带有aioamqp的异步RabbitMQ消费者
我正在尝试使用asyncio / aioamqp编写异步使用者。 我的问题是，回调协程（下面）阻塞。 我将通道设置为执行basic_consume（），并将回调分配为callback（）。 该回调具有“从asyncio.sleep产生”语句（模拟“工作”），该语句从发布者处获取一个整数，并在打印该消息之前休眠该段时间。
如果我发布了两条消息，一条的时间为“10”，紧接着是一条的时间为“1”，我预计第二条消息将首先打印，因为它具有较短的睡眠时间。 相反，回调会阻塞10秒钟，打印第一条消息，然后打印第二条消息。
它显示basic_consume，或回调，阻止某处。 有没有另一种方法可以处理？
@asyncio.coroutine
def callback(body, envelope, properties):
yield from asyncio.sleep(int(body))
print("consumer {} recved {} ({})".format(envelope.consumer_tag, body, envelope.delivery_tag))
@asyncio.coroutine
def receive_log():
try:
transport, protocol = yield from aioamqp.connect('localhost', 5672, login="login", password="password")
except:
print("closed connections")
return
channel = yield from protocol.channel()
exchange_name = 'cloudstack-events'
exchange_name = 'test-async-exchange'
queue_name = 'async-queue-%s' % random.randint(0, 10000)
yield from channel.exchange(exchange_name, 'topic', auto_delete=True, passive=False, durable=False)
yield from asyncio.wait_for(channel.queue(queue_name, durable=False, auto_delete=True), timeout=10)
binding_keys = ['mykey']
for binding_key in binding_keys:
print("binding", binding_key)
yield from asyncio.wait_for(channel.queue_bind(exchange_name=exchange_name,
queue_name=queue_name,
routing_key=binding_key), timeout=10)
print(' [*] Waiting for logs. To exit press CTRL+C')
yield from channel.basic_consume(queue_name, callback=callback)
loop = asyncio.get_event_loop()
loop.create_task(receive_log())
loop.run_forever()
对于那些感兴趣的人，我想出了一个办法来做到这一点。 我不确定这是否是最佳做法，但它正在完成我所需要的。
我不是在回调中做“工作”（在这种情况下，async.sleep），而是在循环中创建一个新任务，并安排一个单独的协程来运行do_work（）。 大概这是工作，因为它释放回调（）立即返回。
我用不同的睡眠定时器在Rabbit中加载了几百个事件，并且在按照下面的代码打印时它们被交错。 所以它似乎工作。 希望这可以帮助别人！
@asyncio.coroutine
def do_work(envelope, body):
yield from asyncio.sleep(int(body))
print("consumer {} recved {} ({})".format(envelope.consumer_tag, body, envelope.delivery_tag))
@asyncio.coroutine
def callback(body, envelope, properties):
loop = asyncio.get_event_loop()
loop.create_task(do_work(envelope, body))
@asyncio.coroutine
def receive_log():
try:
transport, protocol = yield from aioamqp.connect('localhost', 5672, login="login", password="password")
except:
print("closed connections")
return
channel = yield from protocol.channel()
exchange_name = 'cloudstack-events'
exchange_name = 'test-async-exchange'
queue_name = 'async-queue-%s' % random.randint(0, 10000)
yield from channel.exchange(exchange_name, 'topic', auto_delete=True, passive=False, durable=False)
yield from asyncio.wait_for(channel.queue(queue_name, durable=False, auto_delete=True), timeout=10)
binding_keys = ['mykey']
for binding_key in binding_keys:
print("binding", binding_key)
yield from asyncio.wait_for(channel.queue_bind(exchange_name=exchange_name,
queue_name=queue_name,
routing_key=binding_key), timeout=10)
print(' [*] Waiting for logs. To exit press CTRL+C')
yield from channel.basic_consume(queue_name, callback=callback)
loop = asyncio.get_event_loop()
loop.create_task(receive_log())
loop.run_forever()