新增转移终端优化方案

@@ -5,17 +5,15 @@
 # @File : core.py
 # @File : core.py
 # @Software: PyCharm
 # @Software: PyCharm
 import json
 import json
-import threading
 
 
 from tornado import ioloop
 from tornado import ioloop
 
 
 from conf import settings
 from conf import settings
 from core.monitor import Monitor
 from core.monitor import Monitor
-from lib.common import Exchange
+from lib import control_server
 from lib.socket_ import MessageSocketClient
 from lib.socket_ import MessageSocketClient
 
 
-socket = MessageSocketClient()
-monitor = Monitor.bind(socket)
+monitor = Monitor()
 
 
 
 
 def message_receive(message):
 def message_receive(message):
@@ -24,8 +22,6 @@ def message_receive(message):
  monitor.login("[email protected].ru", "nantian888")
  monitor.login("[email protected]", "nantian888")
  elif message == 'exit1':
  elif message == 'exit1':
  monitor.logout('100047151842270')
  monitor.logout('100047151842270')
- elif message == 'uid':
- socket.send({"keys": monitor.members()})
  else:
  else:
  try:
  try:
  req = json.loads(message)
  req = json.loads(message)
@@ -43,12 +39,24 @@ def message_receive(message):
  print(traceback.format_exc())
  print(traceback.format_exc())
 
 
 
 
-def connect_success():
- # Timer(2, monitor._auto_login).start()#连接后发送下
+def server_replace(ws_url):
+ monitor._temp_socket = MessageSocketClient(connect_success, message_receive)
+ monitor._temp_socket.connect(ws_url, reconnect=False)
+
+
+def connect_success(sock):
+ if monitor._temp_socket == sock:
+ sock.auto_reconnet = True
+ monitor._socket.quit() # 旧的断开
+ monitor._temp_socket = None
+
+ monitor.bind(sock)
  monitor._init_server_()
  monitor._init_server_()
+ monitor._socket.new_server_func = lambda: control_server.get_ws_address(monitor._name, sock.ws_url)
+ monitor._socket.server_replace = server_replace
 
 
 
 
 def run():
 def run():
+ socket = MessageSocketClient(connect_success, message_receive)
  socket.connect(**settings.get_server())
  socket.connect(**settings.get_server())
- socket.receive(message_receive).success(connect_success)
  ioloop.IOLoop.instance().start()
  ioloop.IOLoop.instance().start()

@@ -4,6 +4,7 @@
 # @Author : Lemon
 # @Author : Lemon
 # @File : monitor.py
 # @File : monitor.py
 # @Software: PyCharm
 # @Software: PyCharm
+import random
 import time
 import time
 import uuid
 import uuid
 import functools
 import functools
@@ -29,18 +30,15 @@ class Monitor(callback.CallBack):
 
 
  def __init__(self):
  def __init__(self):
  super().__init__()
  super().__init__()
- self._socket = None
+ self._socket = self._temp_socket = None
  self._listenlist = dict()
  self._listenlist = dict()
  self._imei = Config.get('imei', lambda: uuid.uuid1().hex)
  self._imei = Config.get('imei', lambda: uuid.uuid1().hex)
  self._name = Config.get('name', control_server.get_init_name)
  self._name = Config.get('name', control_server.get_init_name)
  self.executor = ThreadPoolExecutor(50, 'task_thread')
  self.executor = ThreadPoolExecutor(50, 'task_thread')
  self.init_config = {}
  self.init_config = {}
 
 
- @classmethod
- def bind(cls, socket):
- self = cls()
+ def bind(self, socket):
  self._socket = socket
  self._socket = socket
- return self
 
 
  def _auto_login(self):
  def _auto_login(self):
  user_list = UserList.query(status=Status.ONLINE)
  user_list = UserList.query(status=Status.ONLINE)
@@ -125,7 +123,7 @@ class Monitor(callback.CallBack):
  'imei': self._imei,
  'imei': self._imei,
  'name': self._name,
  'name': self._name,
  }),
  }),
- "type": "initialize"
+ "type": "initialize",
  }
  }
  self._socket.send(payload)
  self._socket.send(payload)
  self._socket.payload_data = self._heartbeat # 替换心跳数据获取方式
  self._socket.payload_data = self._heartbeat # 替换心跳数据获取方式

@@ -3,9 +3,22 @@
 
 
 import requests
 import requests
 
 
+SERVER = 'http://47.115.27.201:39001'
+
 
 
 def get_init_name():
 def get_init_name():
- r = requests.get('http://47.115.27.201:39001/get_only_int')
+ r = requests.get(SERVER + '/get_only_int')
  num = r.json().get('data', {}).get('only_int', 0)
  num = r.json().get('data', {}).get('only_int', 0)
  assert num, '初始化接口，无法获取终端索引ID'
  assert num, '初始化接口，无法获取终端索引ID'
  return "FBCHAT-{}".format(num)
  return "FBCHAT-{}".format(num)
+
+
+def get_ws_address(name, source_ws):
+ data = {"name": name, "ws": source_ws}
+
+ try:
+ r = requests.post(SERVER + '/post_terminal_info', data, timeout=1)
+ new_ws = r.json().get('data', {}).get('ws', source_ws)
+ except:
+ new_ws = source_ws
+ return new_ws, not new_ws == source_ws

@@ -6,6 +6,9 @@
 # @Software: PyCharm
 # @Software: PyCharm
 import functools
 import functools
 import json
 import json
+import random
+import time
+from threading import Timer
 
 
 from tornado import gen
 from tornado import gen
 from tornado import httpclient
 from tornado import httpclient
@@ -111,20 +114,36 @@ class WebSocketClient(object):
 
 
 class MessageSocketClient(WebSocketClient):
 class MessageSocketClient(WebSocketClient):
 
 
- def __init__(self, connect_timeout=DEFAULT_CONNECT_TIMEOUT,
+ def __init__(self, connect_success=None, message_receive=None, connect_timeout=DEFAULT_CONNECT_TIMEOUT,
  request_timeout=DEFAULT_REQUEST_TIMEOUT):
  request_timeout=DEFAULT_REQUEST_TIMEOUT):
+ self._connect_success = connect_success
+ self._message_receive = message_receive
+
+ self._later_handler = dict()
  self.connect_timeout = connect_timeout
  self.connect_timeout = connect_timeout
  self.request_timeout = request_timeout
  self.request_timeout = request_timeout
  self._io_loop = ioloop.IOLoop.current()
  self._io_loop = ioloop.IOLoop.current()
  self.ws_url = None
  self.ws_url = None
- self.auto_reconnet = False
- self._message_receive = self._connect_success = None
+ self.auto_net = False
+ self.heartbeat_interval_in_secs = 30
 
 
  super(MessageSocketClient, self).__init__(self._io_loop,
  super(MessageSocketClient, self).__init__(self._io_loop,
  self.connect_timeout,
  self.connect_timeout,
  self.request_timeout)
  self.request_timeout)
 
 
- self.heartbeat_interval_in_secs = 30
+ self._later_handler['move_server'] = self._io_loop.call_later(random.randint(30, 60), self.auto_move_server)
+
+ def new_server_func(self):
+ return self.ws_url, False
+
+ def server_replace(self, ws_url):
+ print("需要连接新服务器", ws_url)
+
+ def auto_move_server(self):
+ ws_url, move = self.new_server_func()
+ if move:
+ self.server_replace(ws_url)
+ self._later_handler['move_server'] = self._io_loop.call_later(random.randint(30, 30), self.auto_move_server)
 
 
  def connect(self, url, reconnect=True, reconnect_interval=3):
  def connect(self, url, reconnect=True, reconnect_interval=3):
  self.ws_url = url
  self.ws_url = url
@@ -134,8 +153,9 @@ class MessageSocketClient(WebSocketClient):
 
 
  def on_connection_success(self):
  def on_connection_success(self):
  if self._connect_success:
  if self._connect_success:
- self._connect_success()
- self._io_loop.call_later(self.heartbeat_interval_in_secs, functools.partial(self.sendheartbeat))
+ self._connect_success(self)
+ self._later_handler['heartbeat'] = self._io_loop.call_later(self.heartbeat_interval_in_secs,
+ functools.partial(self.sendheartbeat))
 
 
  def on_connection_close(self, reason):
  def on_connection_close(self, reason):
  print('%s Connection closed reason=%s' % (self.ws_url, reason,))
  print('%s Connection closed reason=%s' % (self.ws_url, reason,))
@@ -143,8 +163,9 @@ class MessageSocketClient(WebSocketClient):
 
 
  def reconnect(self):
  def reconnect(self):
  if not self.is_connected() and self.auto_reconnet:
  if not self.is_connected() and self.auto_reconnet:
- self._io_loop.call_later(self.reconnect_interval,
- super(MessageSocketClient, self).connect, self.ws_url)
+ self._later_handler['reconnect'] = self._io_loop.call_later(self.reconnect_interval,
+ super(MessageSocketClient, self).connect,
+ self.ws_url)
 
 
  def payload_data(self):
  def payload_data(self):
  return {"type": "ping"}
  return {"type": "ping"}
@@ -154,19 +175,42 @@ class MessageSocketClient(WebSocketClient):
  bool = self.send(msg) if msg else False
  bool = self.send(msg) if msg else False
 
 
  if msg is None or bool:
  if msg is None or bool:
- self._io_loop.call_later(self.heartbeat_interval_in_secs,
+ self._later_handler['heartbeat'] = self._io_loop.call_later(self.heartbeat_interval_in_secs,
  functools.partial(self.sendheartbeat))
  functools.partial(self.sendheartbeat))
 
 
- def receive(self, message_receive):
- self._message_receive = message_receive
- return self
-
- def success(self, connect_success):
- self._connect_success = connect_success
- return self
-
  def on_message(self, msg):
  def on_message(self, msg):
  if not self._message_receive:
  if not self._message_receive:
  raise BaseException("未设置接收消息函数->", self.receive)
  raise BaseException("未设置接收消息函数->", self.receive)
  else:
  else:
  self._message_receive(msg)
  self._message_receive(msg)
+
+ def quit(self):
+ self.auto_reconnet = False
+ self.close("quit")
+ for k, handler in self._later_handler.items():
+ self._io_loop.remove_timeout(handler)
+
+
+if __name__ == '__main__':
+
+ def success(sock):
+ global sockccc
+ sockccc = sock
+
+
+ def rec(msg):
+ if msg == 'quit':
+ sockccc.quit()
+
+
+ def interval():
+ if sockccc:
+ sockccc.send({'test': time.time()})
+ Timer(3, interval).start()
+
+
+ s1 = MessageSocketClient()
+ s1.connect('ws://localhost:10000', reconnect=False)
+
+ Timer(2, interval).start()
+ ioloop.IOLoop.current().start()