Asyncio 和 rabbitmq (asynqp)：如何同时从多个队列消费答案

【问题标题】：Asyncio and rabbitmq (asynqp): how to consume from multiple queues concurrentlyAsyncio 和 rabbitmq (asynqp)：如何同时从多个队列消费
【发布时间】：2016-08-11 12:04:06
【问题描述】：

我正在尝试使用 python、asyncio 和 asynqp 同时使用多个队列。

我不明白为什么我的asyncio.sleep() 函数调用没有任何效果。代码不会停在那里。公平地说，我实际上不明白回调是在哪个上下文中执行的，以及我是否可以将控制权交给事件循环（这样asyncio.sleep() 调用才有意义）。

如果我必须在我的process_msg 回调函数中使用aiohttp.ClientSession.get() 函数调用怎么办？我不能这样做，因为它不是协程。必须有一种方法超出我目前对 asyncio 的理解。

#!/usr/bin/env python3

import asyncio
import asynqp


USERS = {'betty', 'bob', 'luis', 'tony'}


def process_msg(msg):
    asyncio.sleep(10)
    print('>> {}'.format(msg.body))
    msg.ack()

async def connect():
    connection = await asynqp.connect(host='dev_queue', virtual_host='asynqp_test')
    channel = await connection.open_channel()
    exchange = await channel.declare_exchange('inboxes', 'direct')

    # we have 10 users. Set up a queue for each of them
    # use different channels to avoid any interference
    # during message consumption, just in case.
    for username in USERS:
        user_channel = await connection.open_channel()
        queue = await user_channel.declare_queue('Inbox_{}'.format(username))
        await queue.bind(exchange, routing_key=username)
        await queue.consume(process_msg)

    # deliver 10 messages to each user
    for username in USERS:
        for msg_idx in range(10):
            msg = asynqp.Message('Msg #{} for {}'.format(msg_idx, username))
            exchange.publish(msg, routing_key=username)


loop = asyncio.get_event_loop()
loop.run_until_complete(connect())
loop.run_forever()

【问题讨论】：

标签： python python-3.x rabbitmq python-asyncio aiohttp

【解决方案1】：

我不明白为什么我的 asyncio.sleep() 函数调用没有任何效果。

因为asyncio.sleep() 返回一个必须与事件循环（或async/await 语义）结合使用的未来对象。

您不能在简单的def 声明中使用await，因为回调是在async/await 上下文之外调用的，该上下文附加到幕后的某个事件循环。换句话说，将回调样式与async/await 样式混合是相当棘手的。

不过，简单的解决方案是将工作安排回事件循环：

async def process_msg(msg):
    await asyncio.sleep(10)
    print('>> {}'.format(msg.body))
    msg.ack()

def _process_msg(msg):
    loop = asyncio.get_event_loop()
    loop.create_task(process_msg(msg))
    # or if loop is always the same one single line is enough
    # asyncio.ensure_future(process_msg(msg))

# some code
await queue.consume(_process_msg)

注意_process_msg 函数中没有递归，即process_msg 的主体在_process_msg 中不执行。一旦控件返回事件循环，就会调用内部的process_msg 函数。

这可以用以下代码概括：

def async_to_callback(coro):
    def callback(*args, **kwargs):
        asyncio.ensure_future(coro(*args, **kwargs))
    return callback

async def process_msg(msg):
    # the body

# some code
await queue.consume(async_to_callback(process_msg))

【讨论】：

【解决方案2】：

请参阅 github 上的 Drizzt1991's response 以获取解决方案。

【讨论】：