异步和无限循环答案

【问题标题】：Asyncio and infinite loop异步和无限循环
【发布时间】：2016-02-02 21:54:16
【问题描述】：

@asyncio.coroutine
    def listener():
        while True:
            message = yield from websocket.recieve_message()
            if message:
                yield from handle(message)

loop = asyncio.get_event_loop()
loop.run_until_complete(listener())

假设我正在使用带有 asyncio 的 websockets。这意味着我收到来自websockets 的消息。当我收到一条消息时，我想处理这条消息，但我的代码丢失了所有异步的东西。因为yield from handle(message) 绝对是阻塞的......我怎么能找到一种方法让它成为非阻塞的？比如，同时处理多条消息。在处理另一条消息之前不必等待消息处理完毕。

谢谢。

【问题讨论】：

通常每个 websocket 需要一个任务来读取，写入 websocket 可能与其他任务异步执行。 handle 也可能是单独的任务。你的代码不完整，所以要得到你真正需要的东西并不容易。
如果这是您正在调用的唯一协程，则侦听器将阻塞，因为它将无限运行（因为 while true 循环）。如果您有另一个协同程序同时运行（具有自己的 yield from 语句），那么 asyncio 将在 yield from 语句之间来回反弹，因此它将不再“阻塞”。

标签： python python-asyncio

【解决方案1】：

如果您不关心句柄消息的返回值，您可以简单地为其创建一个新任务，该任务将在您的 websocket 阅读器旁边的事件循环中运行。这是一个简单的例子：

@asyncio.coroutine
def listener():
    while True:
        message = yield from websocket.recieve_message()
        if message:
            asyncio.ensure_future(handle(message))

ensure_future 将创建一个任务并将其附加到默认事件循环。由于循环已经在运行，它将与您的 websocket 阅读器并行处理。事实上，如果它是一个运行缓慢的 I/O 阻塞任务（例如发送电子邮件），您可以轻松地同时运行几十个句柄（消息）任务。它们在需要时动态创建，并在完成时销毁（开销比生成线程低得多）。

如果您想要更多控制，您可以简单地写入阅读器中的 asyncio.Queue 并拥有一个可以使用队列的固定大小的任务池，这是多线程或多进程编程中的典型模式.

@asyncio.coroutine
def consumer(queue):
    while True:
        message = yield from queue.get()
        yield from handle(message)

@asyncio.coroutine
def listener(queue):
    for i in range(5):
         asyncio.ensure_future(consumer(queue))
    while True:
        message = yield from websocket.recieve_message()
        if message:
            yield from q.put(message)

q = asyncio.Queue()
loop = asyncio.get_event_loop()
loop.run_until_complete(listener(q))

【讨论】：