使用 aioamqp 的异步 RabbitMQ 消费者

Question

我正在尝试使用 asyncio/aioamqp 编写异步使用者。我的问题是，回调协程（下）被阻塞了。我将通道设置为执行 basic_consume()，并将回调分配为 callback()。回调有一个“yield from 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()

Answer 1

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

我没有在回调中执行“工作”（在本例中为 async.sleep），而是在循环中创建一个新任务，并安排一个单独的协同例程来运行 do_work()。大概这是可行的，因为它释放了 callback() 以立即返回。

我在 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()