From 298be0f7e061200e76eeaa8c662b6c116c4f00f4 Mon Sep 17 00:00:00 2001
From: Administrator <admin@example.com>
Date: 星期一, 25 八月 2025 15:39:37 +0800
Subject: [PATCH] 委托记录接口修改
---
test/test_pyzmq.py | 69 +++++++++++++++++++++++++++++++++-
1 files changed, 66 insertions(+), 3 deletions(-)
diff --git a/test/test_pyzmq.py b/test/test_pyzmq.py
index bdc1dc4..f136d8b 100644
--- a/test/test_pyzmq.py
+++ b/test/test_pyzmq.py
@@ -1,8 +1,6 @@
-import json
import multiprocessing
import threading
import time
-
import zmq
@@ -45,7 +43,7 @@
response = socket_client_dict[i].recv_string()
-if __name__ == "__main__":
+if __name__ == "__main__1":
serverProcess = multiprocessing.Process(
target=create_server)
@@ -59,3 +57,68 @@
threading.Thread(target=lambda: send_data(context, i), daemon=True).start()
time.sleep(5)
input()
+
+
+# L2涓婁紶鏁版嵁鍗忚绠＄悊鍣�
+class L2DataUploadProtocolManager:
+
+ # ipchosts IPC鍗忚
+ def __init__(self, ipchosts):
+ self.ipchosts = ipchosts
+ # 鎵�鏈夌殑client
+ self.socket_client_dict = {}
+ # 淇濆瓨浠ｇ爜鍒嗛厤鐨刢lient 鏍煎紡锛歿code:(host, socket)}
+ self.code_socket_client_dict = {}
+ self.rlock = threading.RLock()
+ context = zmq.Context()
+ for host in self.ipchosts:
+ socket = context.socket(zmq.REQ)
+ socket.connect(host)
+ self.socket_client_dict[host] = socket
+
+ # 鑾峰彇
+ def __get_available_ipchost(self):
+ if len(self.code_socket_client_dict) >= len(self.socket_client_dict):
+ raise Exception("鏃犲彲鐢╤ost")
+ used_hosts = set([self.code_socket_client_dict[k][0] for k in self.code_socket_client_dict])
+ for host in self.socket_client_dict:
+ if host not in used_hosts:
+ return host, self.socket_client_dict[host]
+ raise Exception("鏃犲彲鐢╤ost")
+
+ # 鍒嗛厤HOST
+ def distribute_upload_host(self, code):
+ self.rlock.acquire()
+ try:
+ host_info = self.__get_available_ipchost()
+ if host_info:
+ self.code_socket_client_dict[code] = host_info
+ finally:
+ self.rlock.release()
+
+ def release_distributed_upload_host(self, code):
+ self.rlock.acquire()
+ try:
+ if code in self.code_socket_client_dict:
+ self.code_socket_client_dict.pop(code)
+ finally:
+ self.rlock.release()
+
+ def upload_data_as_json(self, code, data):
+ if code not in self.code_socket_client_dict:
+ raise Exception("灏氭湭鍒嗛厤host")
+ host, socket = self.code_socket_client_dict[code]
+ socket.send_json(data)
+ socket.recv_string()
+
+
+if __name__ == "__main__":
+ ipclist = []
+ for i in range(0, 70):
+ ipclist.append(f"ipc://l2order{i}.ipc")
+ manager = L2DataUploadProtocolManager(ipclist)
+ code = "000333"
+ # for i in range(0, 70):
+ # manager.distribute_upload_host(f"{i}" * 6)
+ # manager.distribute_upload_host(code)
+ manager.upload_data_as_json(code, {"test": "test"})
--
Gitblit v1.8.0