feat: 优化cd.py

out_service/cd.py

@@ -3,7 +3,9 @@ import json
 from urllib.parse import urlparse, parse_qs
 import socket
 import traceback
+import threading  # 新增：引入线程锁
 
+# 全局 Socket 连接字典
 sc_all = {
     "192.168.1.220_6000": None,
     "192.168.1.235_6000": None,
@@ -11,6 +13,8 @@ sc_all = {
     "192.168.1.230_6000": None,
 }
 
+sc_lock = threading.Lock()  # 全局锁，保护 sc_all
+
 def get_checksum(body_msg):
     return sum(body_msg) & 0xFF
 
@@ -21,29 +25,26 @@ def handle_bytes(arr):
     if arr[0] != 0xEB or arr[1] != 0x90:
         return "数据头不正确"
 
-
-    # 读取长度信息
-    length_arr = arr[2:4][::-1]  # 反转字节
-    length = int.from_bytes(length_arr, byteorder='little', signed=True)  # 小端格式
+    length_arr = arr[2:4][::-1]
+    length = int.from_bytes(length_arr, byteorder='little', signed=True)
     
-    # 提取内容
-    body = arr[4:4 + length - 3]
+    if len(arr) < length + 4:
+        return f"数据不完整，期望长度:{length+4}，实际长度:{len(arr)}"
 
-    # 校验和检查
+    body = arr[4:4 + length - 3]
     check_sum = get_checksum(body)
     if check_sum != arr[length + 1]:
         return "校验错误"
-        
 
-    # 尾部标识检查
     if arr[length + 2] != 0xFF or arr[length + 3] != 0xFE:
         return "尾错误"
 
-    content = body.decode('utf-8')
-
-    res = json.loads(content)
-    
-    return res[0]
+    try:
+        content = body.decode('utf-8')
+        res = json.loads(content)
+        return res[0]
+    except Exception as e:
+        return f"数据解析错误: {str(e)}"
 
 class JSONRequestHandler(BaseHTTPRequestHandler):
     def ok(self, data):
@@ -56,9 +57,7 @@ class JSONRequestHandler(BaseHTTPRequestHandler):
         self.send_response(400)
         self.send_header('Content-Type', 'application/json; charset=utf-8')
         self.end_headers()
-        data = {
-            "err_msg": err_msg
-        }
+        data = {"err_msg": err_msg}
         self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8'))
 
     def do_GET(self):
@@ -67,50 +66,66 @@ class JSONRequestHandler(BaseHTTPRequestHandler):
         host = query_params.get('host', ['127.0.0.1'])[0]
         port = query_params.get('port', ['6000'])[0]
         addr = f'{host}_{port}'
+        
         if addr not in sc_all:
             self.error(f'{addr} 未找到')
             return
 
         def connect_and_send():
-            sc = sc_all[addr]
+            with sc_lock:  # 加锁，防止竞争
+                sc = sc_all[addr]
+                
             try:
                 if sc is None:
                     sc = socket.socket()
-                    sc.settimeout(5)  # 设置超时
+                    sc.settimeout(5)
                     sc.connect((host, int(port)))
-                    sc_all[addr] = sc
-                # 清空接收缓冲区
-                sc.settimeout(0.1)  # 设置短暂超时
-                for _ in range(5):
+                    with sc_lock:  # 再次加锁，更新 sc_all
+                        sc_all[addr] = sc
+                
+                sc.settimeout(0.5)
+                while True:
                     try:
                         data = sc.recv(65536)
                         if not data:
                             break
                     except (socket.timeout, BlockingIOError):
                         break
-                sc.settimeout(5)  # 恢复原超时设置
+                
+                sc.settimeout(5)
                 sc.sendall(b"R")
-                return sc
+                
+                data = bytearray()
+                while len(data) < 8:
+                    chunk = sc.recv(1024)
+                    if not chunk:
+                        raise ConnectionError("连接中断")
+                    data.extend(chunk)
+                
+                return sc, bytes(data)
+                
             except Exception as e:
                 if sc is not None:
                     try:
                         sc.close()
                     except Exception:
                         pass
-                    sc_all[addr] = None
-                self.error(f'采集器连接失败: {e}')
+                    with sc_lock:  # 加锁，清除无效连接
+                        sc_all[addr] = None
+                self.error(f'采集器通信失败: {e}')
                 print(traceback.format_exc())
-                return None
+                return None, None
 
-        sc = connect_and_send()
-        if sc is None:
+        sc, resp = connect_and_send()
+        if sc is None or resp is None:
             return
-        resp = sc.recv(1024)
+            
         res = handle_bytes(resp)
         if isinstance(res, str):
             self.error(res)
         else:
             self.ok(res)
+
 def run(server_class=HTTPServer, handler_class=JSONRequestHandler, port=2300):
     server_address = ('', port)
     httpd = server_class(server_address, handler_class)