zeromq集成测试

Administrator

2024-02-27 43abbeda445f731d9af0f67a2c5c3e99e8d7b670

zeromq集成测试

 test/test_pyzmq.py

@@ -18,7 +18,7 @@

def create_server():
    context = zmq.Context()
    for i in range(5):
    for i in range(70):
        threading.Thread(target=lambda: __create_server(context, i), daemon=True).start()
    while True:
        time.sleep(5)
@@ -46,7 +46,8 @@
    time.sleep(5)
    context = zmq.Context()
    while True:
        for i in range(5):
        print("=======准备发送消息========")
        for i in range(70):
            threading.Thread(target=lambda: send_data(context, i), daemon=True).start()
        time.sleep(5)
    input()