Python - 如何使用 asyncio 同时运行多个协程？

Question

我正在使用 websockets 库在 Python 3.4 中创建一个 websocket 服务器。这是一个简单的回显服务器：

 import asyncio
import websockets

@asyncio.coroutine
def connection_handler(websocket, path):
    while True:
        msg = yield from websocket.recv()
        if msg is None:  # connection lost
            break
        yield from websocket.send(msg)

start_server = websockets.serve(connection_handler, 'localhost', 8000)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

假设我们 - 另外 - 想要在某些事件发生时向客户端发送消息。为简单起见，让我们每 60 秒定期发送一条消息。我们该怎么做？我的意思是，因为 connection_handler 一直在等待传入的消息，服务器只有在收到来自客户端的消息 后 才能采取行动，对吗？我在这里错过了什么？

也许这种情况需要一个基于事件/回调的框架而不是一个基于协程的框架？ 龙卷风？

原文由 weatherfrog 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

TL;DR 使用 asyncio.ensure_future() 同时运行多个协程。

---

也许这种情况需要一个基于事件/回调的框架而不是一个基于协程的框架？龙卷风？

不，您不需要任何其他框架。异步应用程序与同步应用程序的整体思想是它在等待结果时不会阻塞。它是如何实现的，使用协程或回调并不重要。

我的意思是，因为 connection_handler 一直在等待传入的消息，所以服务器只能在收到来自客户端的消息后才能采取行动，对吧？我在这里错过了什么？

在同步应用程序中，您将编写类似 msg = websocket.recv() 的内容，这将阻止整个应用程序，直到您收到消息（如您所述）。但是在异步应用中就完全不同了。

当你做 msg = yield from websocket.recv() 你说这样的话：暂停执行 connection_handler() 直到 websocket.recv() 会产生一些东西。在协程中使用 yield from 将控制返回给事件循环，因此可以执行其他一些代码，同时我们正在等待 websocket.recv() 的结果。请参阅 文档 以更好地了解协同程序的工作原理。

假设我们 - 另外 - 想要在某些事件发生时向客户端发送消息。为简单起见，让我们每 60 秒定期发送一条消息。我们该怎么做？

在执行 启动事件循环 的阻塞调用之前，您可以使用 asyncio.async() 运行任意数量的协程。

 import asyncio

import websockets

# here we'll store all active connections to use for sending periodic messages
connections = []

@asyncio.coroutine
def connection_handler(connection, path):
    connections.append(connection)  # add connection to pool
    while True:
        msg = yield from connection.recv()
        if msg is None:  # connection lost
            connections.remove(connection)  # remove connection from pool, when client disconnects
            break
        else:
            print('< {}'.format(msg))
        yield from connection.send(msg)
        print('> {}'.format(msg))

@asyncio.coroutine
def send_periodically():
    while True:
        yield from asyncio.sleep(5)  # switch to other code and continue execution in 5 seconds
        for connection in connections:
            print('> Periodic event happened.')
            yield from connection.send('Periodic event happened.')  # send message to each connected client

start_server = websockets.serve(connection_handler, 'localhost', 8000)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.async(send_periodically())  # before blocking call we schedule our coroutine for sending periodic messages
asyncio.get_event_loop().run_forever()

这是一个示例客户端实现。它要求您输入名称，从回显服务器接收它，等待来自服务器的另外两条消息（这是我们的定期消息）并关闭连接。

 import asyncio

import websockets

@asyncio.coroutine
def hello():
    connection = yield from websockets.connect('ws://localhost:8000/')
    name = input("What's your name? ")
    yield from connection.send(name)
    print("> {}".format(name))
    for _ in range(3):
        msg = yield from connection.recv()
        print("< {}".format(msg))

    yield from connection.close()

asyncio.get_event_loop().run_until_complete(hello())

要点：

1. 在 Python 3.4.4 asyncio.async() 重命名为 asyncio.ensure_future() 。
2. 有用于安排 延迟调用 的特殊方法，但它们不适用于协程。

原文由 Yaroslav Admin 发布，翻译遵循 CC BY-SA 3.0 许可协议

Answer 2

我很惊讶 gather 没有被提及。

来自 Python 文档：

 import asyncio

async def factorial(name, number):
    f = 1
    for i in range(2, number + 1):
        print(f"Task {name}: Compute factorial({i})...")
        await asyncio.sleep(1)
        f *= i
    print(f"Task {name}: factorial({number}) = {f}")

async def main():
    # Schedule three calls *concurrently*:
    await asyncio.gather(
        factorial("A", 2),
        factorial("B", 3),
        factorial("C", 4),
    )

asyncio.run(main())

# Expected output:
#
#     Task A: Compute factorial(2)...
#     Task B: Compute factorial(2)...
#     Task C: Compute factorial(2)...
#     Task A: factorial(2) = 2
#     Task B: Compute factorial(3)...
#     Task C: Compute factorial(3)...
#     Task B: factorial(3) = 6
#     Task C: Compute factorial(4)...
#     Task C: factorial(4) = 24

原文由 Cyril N. 发布，翻译遵循 CC BY-SA 4.0 许可协议