zeromq集成测试

Administrator

2024-02-27 4fd8f66e1c96abbefdfef73b63084ff60872bb70

zeromq集成测试

 test/test_pyzmq.py

@@ -11,8 +11,9 @@
    socket.bind(f"ipc://order_{index}.ipc")
    print("绑定成功", index)
    while True:
        msg = socket.recv_string()
        print(f"Got Msg({index}-{time.time()}): {msg}")
        msg = socket.recv_json()
        use_time = (time.time() - msg['time'])*1000
        print(f"Got Msg({index}-{time.time()}): 耗时：{use_time}")
        socket.send_string("SUCCESS")


@@ -33,7 +34,7 @@
        socket.connect(f"ipc://order_{i}.ipc")
        socket_client_dict[i] = socket
    print(f"开始发送消息({i})：")
    socket_client_dict[i].send_string(json.dumps({'msg': i, "time": time.time()}))
    socket_client_dict[i].send_json({'msg': i, "time": time.time()})
    response = socket_client_dict[i].recv_string()