服务器和代理服务器之间的连接添加心跳机制
This commit is contained in:
11
server.py
11
server.py
@@ -21,13 +21,13 @@ SERVER_PORT = 5345
|
||||
|
||||
|
||||
# 定义个函数,使其专门重复处理客户的请求数据(也就是重复接受一个用户的消息并且重复回答,直到用户选择下线)
|
||||
def dispose_client_request(tcp_client_1,tcp_addr):
|
||||
def dispose_client_request(tcp_client,tcp_addr):
|
||||
myprint(f"客户端:{tcp_addr} 已连接")
|
||||
recv_data=bytearray()
|
||||
target =tg.tcp_target(tcp_client_1)
|
||||
target =tg.tcp_target(tcp_client)
|
||||
while True:
|
||||
try:
|
||||
recv = tcp_client_1.recv(4096)
|
||||
recv = tcp_client.recv(4096)
|
||||
except Exception as e:
|
||||
myprint(str(e))
|
||||
break
|
||||
@@ -44,7 +44,6 @@ def dispose_client_request(tcp_client_1,tcp_addr):
|
||||
break
|
||||
myprint(f"客户端:{tcp_addr} 已下线")
|
||||
target.close()
|
||||
tcp_client_1.close()
|
||||
|
||||
|
||||
|
||||
@@ -57,8 +56,8 @@ def start_service():
|
||||
tcp_server.listen(128)
|
||||
|
||||
while True:
|
||||
tcp_client_1 , tcp_client_address = tcp_server.accept()
|
||||
thd = threading.Thread(target = dispose_client_request, args = (tcp_client_1,tcp_client_address))
|
||||
tcp_client , tcp_client_address = tcp_server.accept()
|
||||
thd = threading.Thread(target = dispose_client_request, args = (tcp_client,tcp_client_address))
|
||||
# 设置守护主线程 即如果主线程结束了 那子线程中也都销毁了 防止主线程无法退出
|
||||
# thd.setDaemon(True)
|
||||
thd.start()
|
||||
|
17
target.py
17
target.py
@@ -28,13 +28,13 @@ from log import log_init
|
||||
|
||||
# 这里是公共变量,所有tcp_target对象都可以访问
|
||||
# 只支持一对一连接
|
||||
_tcp_server=None
|
||||
_tcp_client=None
|
||||
_tcp_server:socket.socket=None
|
||||
_tcp_client:socket.socket=None
|
||||
|
||||
|
||||
|
||||
class tcp_target(object):
|
||||
def __init__(self,tcp_handler:socket):
|
||||
def __init__(self,tcp_handler:socket.socket):
|
||||
self.handler=tcp_handler
|
||||
self.state="idle"
|
||||
def recv_handler(self,recv_data:bytearray):
|
||||
@@ -60,12 +60,17 @@ class tcp_target(object):
|
||||
if(_tcp_server is not None):
|
||||
_tcp_server.close()
|
||||
_tcp_server=self.handler
|
||||
elif(j["option"]=="keeplive"):
|
||||
# 回复心跳数据
|
||||
cmd={'device':'proxy','option':'keeplive'}
|
||||
data=pc.encode(json.dumps(cmd).encode('utf-8'),b'default')
|
||||
if(_tcp_server is not None):
|
||||
_tcp_server.send(data)
|
||||
else:
|
||||
# 其他消息原样发送到客户端
|
||||
if(_tcp_client is not None):
|
||||
_tcp_client.send(recv_data)
|
||||
# 通知服务器和客户端断开连接
|
||||
# 这里只进行数据的收发处理,不执行端口的close操作
|
||||
def close(self):
|
||||
cmd={'device':'proxy','option':'close'}
|
||||
data=pc.encode(json.dumps(cmd).encode('utf-8'),b'default')
|
||||
@@ -73,13 +78,17 @@ class tcp_target(object):
|
||||
if(_tcp_client is not None):
|
||||
try:
|
||||
_tcp_client.send(data)
|
||||
_tcp_client.close()
|
||||
except Exception as e:
|
||||
myprint("target_close:",str(e))
|
||||
_tcp_client=None
|
||||
if(_tcp_server is not None):
|
||||
try:
|
||||
_tcp_server.send(data)
|
||||
_tcp_server.close()
|
||||
except Exception as e:
|
||||
myprint("target_close:",str(e))
|
||||
_tcp_server=None
|
||||
|
||||
|
||||
|
||||
|
@@ -31,8 +31,8 @@ _local_client=[]
|
||||
|
||||
|
||||
# LOCAL_SERVER_IP = ("192.168.3.166",80)
|
||||
# LOCAL_SERVER_IP = ("192.168.3.167",22)
|
||||
LOCAL_SERVER_IP = ("192.168.3.174",22)
|
||||
LOCAL_SERVER_IP = ("192.168.1.40",22)
|
||||
# LOCAL_SERVER_IP = ("192.168.3.174",22)
|
||||
|
||||
|
||||
|
||||
@@ -69,16 +69,27 @@ def close_all():
|
||||
myprint('连接列表已清空')
|
||||
|
||||
|
||||
|
||||
_tick_start:float=0
|
||||
_tick_end:float=0
|
||||
|
||||
# 定时任务
|
||||
def remote_keeplive():
|
||||
cmd={'device':'server','option':'login'}
|
||||
global _tick_start
|
||||
global _tick_end
|
||||
cmd={'device':'server','option':'keeplive'}
|
||||
data=pc.encode(json.dumps(cmd).encode('utf-8'),b'default')
|
||||
try:
|
||||
_remote_client.send(data)
|
||||
except Exception as e:
|
||||
myprint(str(e))
|
||||
_tick_end=time.perf_counter()
|
||||
# 超过15秒没收到数据则自动断开
|
||||
if(_tick_end-_tick_start>15):
|
||||
myprint("长时间没收到代理服务器的数据回复,主动断开连接")
|
||||
try:
|
||||
_remote_client.close()
|
||||
except Exception as e:
|
||||
myprint(str(e))
|
||||
|
||||
|
||||
|
||||
@@ -122,9 +133,10 @@ def local_client_handler(tcp_server:socket.socket,ip,port):
|
||||
|
||||
|
||||
# 远端数据处理,解包,把负载数据发送到本地服务器
|
||||
def remote_client_handler(tcp_client_1:socket.socket):
|
||||
def remote_client_handler(tcp_client:socket.socket):
|
||||
global _remote_client
|
||||
global _local_client
|
||||
global _tick_start
|
||||
myprint("已连接代理服务器")
|
||||
timer=threading.Timer(5,remote_keeplive,())
|
||||
cmd={'device':'server','option':'login'}
|
||||
@@ -132,15 +144,18 @@ def remote_client_handler(tcp_client_1:socket.socket):
|
||||
_remote_client.send(data)
|
||||
timer.start()
|
||||
recv_data=bytearray()
|
||||
_tick_start=time.perf_counter()
|
||||
while True:
|
||||
try:
|
||||
recv = tcp_client_1.recv(4096)
|
||||
recv = tcp_client.recv(4096)
|
||||
except Exception as e:
|
||||
myprint("remote:",str(e))
|
||||
break
|
||||
if recv:
|
||||
timer.cancel()
|
||||
timer=threading.Timer(5,remote_keeplive,())
|
||||
timer.start()
|
||||
_tick_start=time.perf_counter()
|
||||
recv_data+=recv
|
||||
while True:
|
||||
start=recv_data.find(b'\xff')
|
||||
@@ -168,6 +183,8 @@ def remote_client_handler(tcp_client_1:socket.socket):
|
||||
if(j['option']=='close'):
|
||||
myprint("收到代理服务器的断开通知")
|
||||
close_all()
|
||||
elif(j["option"]=='keeplive'):
|
||||
myprint("收到代理服务器的心跳数据")
|
||||
except Exception as e:
|
||||
myprint(str(e))
|
||||
recv_data=recv_data[end+1:]
|
||||
@@ -175,7 +192,10 @@ def remote_client_handler(tcp_client_1:socket.socket):
|
||||
break
|
||||
timer.cancel()
|
||||
timer.join()
|
||||
tcp_client_1.close()
|
||||
try:
|
||||
tcp_client.close()
|
||||
except Exception as e:
|
||||
myprint(str(e))
|
||||
myprint("与代理服务器的连接已断开")
|
||||
close_all()
|
||||
|
||||
|
Reference in New Issue
Block a user