使用 aioamqp 的异步 RabbitMQ 消费者

Question

我正在尝试使用 asyncio/aioamqp 编写异步使用者。我的问题是，回调协程（下方）正在阻塞。我将通道设置为执行 basic_consume()，并将回调分配为 callback()。回调有一个 "yield from asyncio.sleep" 语句（模拟 "work"），它从发布者那里获取一个整数并在打印消息之前休眠一段时间。

如果我发布了两条消息，一条时间为“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()

Answer 1

对于那些感兴趣的人，我想出了一个方法来做到这一点。我不确定这是否是最佳做法，但它正在完成我的需要。

我没有在回调中执行 "work"（在本例中为 async.sleep），而是在循环中创建了一个新任务，并为 [=21= 安排了一个单独的协程] do_work()。大概这是可行的，因为它立即将 callback() 释放到 return。

我用不同的睡眠定时器在 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()

使用 aioamqp 的异步 RabbitMQ 消费者

Asynchronous RabbitMQ consumer with aioamqp

python

asynchronous

rabbitmq

python-asyncio