如何在带有异步功能的pyzmq中使用REQ和REP？

Question

我正在尝试在python3.5中使用pyzmq和asyncio实现异步客户端和服务器。我使用了zmq提供的asyncio库。下面是我的client（requester.py）和server（responder.py）的代码。我的要求是仅使用REQ和REP zmq套接字来实现异步客户端服务器。

requester.py：

import asyncio
import zmq
import zmq.asyncio

async def receive():
        message = await socket.recv()
        print("Received reply ", "[", message, "]")
        return message

async def send(i):
        print("Sending request ", i,"...")
        request = "Hello:" + str(i)
        await socket.send(request.encode('utf-8'))
        print("sent:",i)

async def main_loop_num(i):
        await send(i)
        #  Get the reply.
        message = await receive()
        print("Message :", message)

async def main():
        await asyncio.gather(*(main_loop_num(i) for i in range(1,10)))                

port = 5556
context = zmq.asyncio.Context.instance()

socket = context.socket(zmq.REQ)
socket.connect("tcp://localhost:%d" % port)
asyncio.get_event_loop().run_until_complete(asyncio.wait([main()]))

responder.py：

import asyncio
import zmq
import zmq.asyncio

async def receive():
    message = await socket.recv()
    print("Received message:", message)
    await asyncio.sleep(10)
    print("Sleep complete")
    return message

async def main_loop():
    while True:
        message = await receive()
        print("back to main loop")
        await socket.send(("World from %d" % port).encode('utf-8'))
        print("sent back")

port = 5556
context = zmq.asyncio.Context.instance()

socket = context.socket(zmq.REP)
socket.bind("tcp://*:%d" % port)
asyncio.get_event_loop().run_until_complete(asyncio.wait([main_loop()]))

我得到的输出是：

requester.py：

Sending request  5 ...
sent: 5
Sending request  6 ...
Sending request  1 ...
Sending request  7 ...
Sending request  2 ...
Sending request  8 ...
Sending request  3 ...
Sending request  9 ...
Sending request  4 ...

responder.py：

Received message: b'Hello:5'
Sleep complete
back to main loop
sent back

从输出中，我假设请求者已发送多个请求，但只有第一个已到达响应者。同样，响应者针对第一个请求发送的响应甚至没有返回到请求者。为什么会这样？我尽可能地使用了异步方法，但send（）和recv（）方法仍然不异步。是否可以在不使用路由器，经销商等任何其他套接字的情况下进行异步req-rep？

Answer 1

ZMQ的REQ-REP套接字期望一个请求的严格顺序-一个答复-一个请求-一个答复-...

您的requester.py并行启动所有10个请求：

await asyncio.gather(*(main_loop_num(i) for i in range(1,10)))

[在发送第二个请求时，ZMQ对此表示抱怨：

zmq.error.ZMQError: Operation cannot be accomplished in current state

尝试更改您的主要功能以一次发送一个请求：

async def main():
    for i in range(1, 10):
        await main_loop_num(i)

---

如果需要并行发送多个请求，则不能使用REQ-REP套接字对，例如，可以使用DEALER-REP套接字对。