zcdsj
/
factory
5
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: 优化get_tyy_data

This commit is contained in:
caoqianming 2025-06-10 12:42:02 +08:00
parent 5e0c515929
commit 8efd28633c
1 changed files with 108 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

144
apps/em/cd.py
View File

@ -5,6 +5,10 @@ import time
from django.core.cache import cache from django.core.cache import cache
from apps.utils.thread import MyThread from apps.utils.thread import MyThread
import struct import struct
import uuid
import logging
myLogger = logging.getLogger('log')
def get_checksum(body_msg): def get_checksum(body_msg):
return sum(body_msg) & 0xFF return sum(body_msg) & 0xFF
@ -19,7 +23,7 @@ def handle_bytes(arr):
# 读取长度信息 # 读取长度信息
length_arr = arr[2:4][::-1] # 反转字节 length_arr = arr[2:4][::-1] # 反转字节
length = struct.unpack('<H', bytes(length_arr))[0] # 小端格式 length = int.from_bytes(length_arr, byteorder='little', signed=True) # 小端格式
# 提取内容 # 提取内容
body = arr[4:4 + length - 3] body = arr[4:4 + length - 3]
@ -40,8 +44,9 @@ def handle_bytes(arr):
return res[0] return res[0]
def get_tyy_data_t(host, port): def get_tyy_data_t(host, port, tid):
cd_thread_key = f"cd_thread_{host}_{port}" cd_thread_key_id = f"cd_thread_{host}_{port}_id"
cd_thread_key_val = f"cd_thread_{host}_{port}_val"
sc = None sc = None
def connect_and_send(retry=1): def connect_and_send(retry=1):
nonlocal sc nonlocal sc
@ -51,55 +56,62 @@ def get_tyy_data_t(host, port):
sc.connect((host, int(port))) sc.connect((host, int(port)))
sc.sendall(b"R") sc.sendall(b"R")
except BrokenPipeError: except BrokenPipeError:
sc = None
if retry > 0: if retry > 0:
connect_and_send(retry-1) connect_and_send(retry-1)
else:
if sc:
try:
sc.close()
except Exception:
pass
sc = None
except OSError as e: except OSError as e:
sc = None sc = None
cache.set(cd_thread_key, {"err_msg": f"采集器连接失败-{str(e)}"}) cache.set(cd_thread_key_val, {"err_msg": f"采集器连接失败-{str(e)}"})
except ConnectionResetError: except ConnectionResetError:
sc = None sc = None
cache.set(cd_thread_key, {"err_msg": "采集器重置了连接"}) cache.set(cd_thread_key_val, {"err_msg": "采集器重置了连接"})
except socket.timeout: except socket.timeout:
sc = None sc = None
cache.set(cd_thread_key, {"err_msg": "采集器连接超时"}) cache.set(cd_thread_key_val, {"err_msg": "采集器连接超时"})
except Exception as e: except Exception as e:
sc = None sc = None
cache.set(cd_thread_key, {"err_msg": f"采集器连接失败-{str(e)}"}) cache.set(cd_thread_key_val, {"err_msg": f"采集器连接失败-{str(e)}"})
while True: while cache.get(cd_thread_key_id) == tid:
cd_thread_val = cache.get(cd_thread_key, default=None) if cache.get(cd_thread_key_val) == "get":
if cd_thread_val is None: cache.set(cd_thread_key_val, "working")
if sc:
try:
sc.close()
except Exception:
pass
break
elif cd_thread_val == "get":
cache.set(cd_thread_key, "working")
connect_and_send() connect_and_send()
if sc is None: if sc is None:
continue continue
resp = sc.recv(1024) resp = sc.recv(1024)
res = handle_bytes(resp) res = handle_bytes(resp)
if isinstance(res, str): if isinstance(res, str):
cache.set(cd_thread_key, {"err_msg": f'采集器返回数据错误-{res}'}) cache.set(cd_thread_key_val, {"err_msg": f'采集器返回数据错误-{res}'})
elif not res:
cache.set(cd_thread_key_val, {"err_msg": f"采集器返回数据为空-{str(res)}"})
else: else:
cache.set(cd_thread_key, res) myLogger.info(f"采集器返回数据-{res}")
cache.set(cd_thread_key_val, res)
time.sleep(0.3) time.sleep(0.3)
def get_tyy_data(*args, sleep=0): if sc:
if sleep > 0: try:
time.sleep(sleep) sc.close()
except Exception:
pass
def get_tyy_data_2(*args, retry=1):
host, port = args[0], int(args[1]) host, port = args[0], int(args[1])
cd_thread_key = f"cd_thread_{host}_{port}" cd_thread_key_id = f"cd_thread_{host}_{port}_id"
cd_thread_val = cache.get(cd_thread_key, default=None) cd_thread_key_val = f"cd_thread_{host}_{port}_val"
if cd_thread_val is None: cd_thread_val_id = cache.get(cd_thread_key_id, default=None)
cache.set(cd_thread_key, "start") if cd_thread_val_id is None:
cd_thread = MyThread(target=get_tyy_data_t, args=(host, port), daemon=True) tid = uuid.uuid4()
cache.set(cd_thread_key_id, tid, timeout=10800)
cd_thread = MyThread(target=get_tyy_data_t, args=(host, port, tid), daemon=True)
cd_thread.start() cd_thread.start()
cache.set(cd_thread_key, "get") cache.set(cd_thread_key_val, "get")
num = 0 num = 0
get_val = False get_val = False
@ -107,7 +119,7 @@ def get_tyy_data(*args, sleep=0):
num += 1 num += 1
if num > 8: if num > 8:
break break
val = cache.get(cd_thread_key) val = cache.get(cd_thread_key_val)
if isinstance(val, dict): if isinstance(val, dict):
get_val = True get_val = True
if "err_msg" in val: if "err_msg" in val:
@ -115,10 +127,70 @@ def get_tyy_data(*args, sleep=0):
return val return val
time.sleep(0.3) time.sleep(0.3)
if not get_val: if not get_val and retry > 0:
cache.set(cd_thread_key, None) cache.set(cd_thread_key_id, None)
get_tyy_data(*args, sleep=2) get_tyy_data_2(*args, retry=retry-1)
if __name__ == '__main__': sc_all = {}
print(get_tyy_data())
def get_tyy_data(*args):
host, port = args[0], int(args[1])
global sc_all
sc = None
def connect_and_send(retry=1):
nonlocal sc
sc = sc_all.get(f"{host}_{port}", None)
try:
if sc is None:
sc = socket.socket()
sc.settimeout(5) # 设置超时
sc.connect((host, port))
sc_all[f"{host}_{port}"] = sc
sc.sendall(b"R")
except BrokenPipeError:
if retry > 0:
if sc:
try:
sc.close()
except Exception:
pass
sc_all.pop(f"{host}_{port}", None)
return connect_and_send(retry-1)
else:
if sc:
try:
sc.close()
except Exception:
pass
sc_all.pop(f"{host}_{port}", None)
sc = None
raise ParseError("采集器连接失败-管道重置")
except OSError as e:
if sc:
try:
sc.close()
except Exception:
pass
sc_all.pop(f"{host}_{port}", None)
sc = None
raise ParseError(f"采集器连接失败-{str(e)}")
except Exception as e:
if sc:
try:
sc.close()
except Exception:
pass
sc_all.pop(f"{host}_{port}", None)
sc = None
raise ParseError(f"采集器连接失败-{str(e)}")
connect_and_send()
resp = sc.recv(1024)
res = handle_bytes(resp)
# myLogger.error(res)
if isinstance(res, str):
raise ParseError(f'采集器返回数据错误-{res}')
else:
return res