3

我正在编写简单的生产者/消费者程序。

import zmq

@asyncio.coroutine
def receive_data(future,s):
        print("begin to recv sth from.....socket"
        my_data = s.recv()
        future.set_result(my_data)

@asyncio.coroutine
def producer(loop,q,s):
        while True:
                future = asyncio.Future()
                yield from receive_data(future,s)
                data = str(future.result())
                yield from q.put(data)
@asyncio.coroutine
def consumer(loop,q):
       while True:
          a = yield from q.get()
          print("i am get..."+str(a)+"..."+str(type(a)))  
loop = asyncio.get_event_loop()

c = zmq.Context()
s = c.socket(zmq.REP)
s.bind('tcp://127.0.0.1:5515')

q = asyncio.Queue()
tasks=[asyncio.Task(producer(loop,q,s)),asyncio.Task(comsumer(loop,q))]
loop.run_until_complete(asyncio.gather(*tasks))
loop.close()
s.close()

消费者似乎没有机会执行。

套接字每 500 毫秒接收一次数据，因此当yield from在 receive_data 函数中挂起生产者协程时，消费者协程将打印信息。

这有什么可以解释的？

4

3 回答 3

3

s.recv()正在阻塞呼叫，所以receive_data挂起直到新的 ZMQ 消息到达。

这会阻塞事件循环，消费者没有机会执行自己。

如果没有可用数据，您可以将zmq.NOBLOCK标志传递给.recv并调用asyncio.sleep(0)，以使 eventloop 有机会迭代其他准备好的任务。

或者只是使用aiozmq库:)

于 2014-11-28T13:48:27.563 回答
1

您正在混合同步和异步调用，结果将是同步的。如果你想继续使用 asyncio，你应该定义一个异步上下文c = zmq.asyncio.context()并使用 ROUTER socket s = c.socket(zmq.ROUTER)。然后，按照 asyncio 语法，你应该从recv_multipart()让你my_data = s.recv()变成my_data = yield from s.recv_multipart().

于 2018-02-26T09:49:38.223 回答
0

以下是需要发生的事情的指南：

  • 你应该使用Contextand ZMQEventLoopfromzmq.asyncio
  • 你应该asyncio通过调用来使用 zmq 循环set_event_loop()
  • 仅用于测试目的，我已切换到 ROUTER_RAW 套接字

工作示例：

import asyncio
import zmq
from zmq.asyncio import Context, ZMQEventLoop

async def receive_data(s):
        data = await s.recv()
        print('receive_data', data)
        return data

async def producer(q, s):
        while True:
                data = await receive_data(s)
                await q.put(data)

async def consumer(q):
       while True:
          a = await q.get()
          print('i got... {} ... {}'.format(a, type(a)))  

loop = ZMQEventLoop()
asyncio.set_event_loop(loop)

c = Context()
s = c.socket(zmq.ROUTER)
s.setsockopt(zmq.ROUTER_RAW, 1)
s.bind('tcp://127.0.0.1:5515')

q = asyncio.Queue()
tasks=[producer(q, s), consumer(q)]
loop.run_until_complete(asyncio.gather(*tasks))
loop.close()
s.close()

您可以使用 telnet 使用 ROUTER_RAW 对其进行测试：

$ telnet localhost 5515
Trying ::1...
telnet: connect to address ::1: Connection refused
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
test
abcd1234
^]
telnet> Connection closed.
$

应用程序对此的响应是：

receive_data b'\x00\xa3\x8e\x1f)'
receive_data b''
i got... b'\x00\xa3\x8e\x1f)' ... <class 'bytes'>
i got... b'' ... <class 'bytes'>
receive_data b'\x00\xa3\x8e\x1f)'
receive_data b'test\r\n'
i got... b'\x00\xa3\x8e\x1f)' ... <class 'bytes'>
i got... b'test\r\n' ... <class 'bytes'>
receive_data b'\x00\xa3\x8e\x1f)'
receive_data b'abcd1234\r\n'
i got... b'\x00\xa3\x8e\x1f)' ... <class 'bytes'>
i got... b'abcd1234\r\n' ... <class 'bytes'>
receive_data b'\x00\xa3\x8e\x1f)'
receive_data b''
i got... b'\x00\xa3\x8e\x1f)' ... <class 'bytes'>
i got... b'' ... <class 'bytes'>
于 2018-03-10T22:18:52.850 回答