From 35b717df5e2498b00aac51672c50873353837383 Mon Sep 17 00:00:00 2001
From: Administrator <admin@example.com>
Date: 星期二, 27 二月 2024 16:48:55 +0800
Subject: [PATCH] zeromq集成测试
---
test/test_pyzmq.py | 68 +++++++++++++++++++++++++++++++--
1 files changed, 63 insertions(+), 5 deletions(-)
diff --git a/test/test_pyzmq.py b/test/test_pyzmq.py
index 9a1a2ba..8bf0a7b 100644
--- a/test/test_pyzmq.py
+++ b/test/test_pyzmq.py
@@ -1,8 +1,6 @@
-import json
import multiprocessing
import threading
import time
-import huaxin_client
import zmq
@@ -61,7 +59,67 @@
input()
-if __name__ == "__main__":
- huaxin_client.l2_data_manager.test()
+# L2涓婁紶鏁版嵁鍗忚绠＄悊鍣�
+class L2DataUploadProtocolManager:
- pass
\ No newline at end of file
+ # ipchosts IPC鍗忚
+ def __init__(self, ipchosts):
+ self.ipchosts = ipchosts
+ # 鎵�鏈夌殑client
+ self.socket_client_dict = {}
+ # 淇濆瓨浠ｇ爜鍒嗛厤鐨刢lient 鏍煎紡锛歿code:(host, socket)}
+ self.code_socket_client_dict = {}
+ self.lock = threading.Lock()
+ 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):
+ try:
+ 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")
+ finally:
+ self.lock.release()
+
+ # 鍒嗛厤HOST
+ def distribute_upload_host(self, code):
+ self.lock.acquire()
+ try:
+ host_info = self.__get_available_ipchost()
+ if host_info:
+ self.code_socket_client_dict[code] = host_info
+ finally:
+ self.lock.release()
+
+ def release_distributed_upload_host(self, code):
+ self.lock.acquire()
+ try:
+ if code in self.code_socket_client_dict:
+ self.code_socket_client_dict.pop(code)
+ finally:
+ self.lock.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"
+ manager.distribute_upload_host(code)
+ manager.upload_data_as_json(code, {"test": "test"})
--
Gitblit v1.8.0