如何使用 asyncio 同时运行多个协程?

问题描述 投票:0回答:4

我正在使用

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
一直在等待传入的消息,所以服务器只能在 收到来自客户端的消息后才采取行动,对吧?我在这里错过了什么?

也许这个场景需要一个基于事件/回调的框架而不是一个基于协程的框架? 龙卷风?

python python-3.x websocket python-asyncio coroutine
4个回答
41
投票

你可以使用

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

35
投票

TL;DR 您可以使用

asyncio.create_task()
同时运行多个协程。


也许这个场景需要一个基于事件/回调的框架而不是一个基于协程的框架?龙卷风?

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

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

在同步应用程序中,您将编写类似

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

当你做

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

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

您可以使用

asyncio.async()
运行任意数量的协程,然后再执行阻塞调用starting event loop

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()
    ,在 Python 3.7 中添加了
    asyncio.create_task()
    ,比
    asyncio.ensure_future()
    更受欢迎。

  2. 调度延迟调用有特殊的方法,但它们不适用于协程。


9
投票

同样的问题,在我看到这里的完美示例之前很难得到解决方案:http://websockets.readthedocs.io/en/stable/intro.html#both

 done, pending = await asyncio.wait(
        [listener_task, producer_task],
        return_when=asyncio.FIRST_COMPLETED)  # Important

这样,我就可以处理心跳、redis订阅等多协程任务了


2
投票

如果您使用的是Python 3.7及更高版本,您可以按如下方式使用

asyncio.gather()
asyncio.run()

import asyncio

async def coro1():
    for i in range(1, 6):
        print(i)
        await asyncio.sleep(0)  # switches task every one iteration.

async def coro2():
    for i in range(1, 6):
        print(i * 10)
        await asyncio.sleep(0)  # switches task every one iteration.

async def main():
    await asyncio.gather(
        coro1(),
        coro2(),
    )

asyncio.run(main())


## Or instead of defining the main async function:

futures = [coro1(), coro2()]
await asyncio.gather(*futures)

使用Python 3.11

中的新
asyncio.TaskGroup你不需要
.gather()

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(coro1())
        tg.create_task(coro2())

asyncio.run(main())

否则,如果您使用的是 Python 3.6 或 3.5,请执行以下操作以获得您应该处理循环的相同结果:

import asyncio

async def coro1():
    for i in range(1, 6):
        print(i)
        await asyncio.sleep(0)  # switches task every one iteration.

async def coro2():
    for i in range(1, 6):
        print(i * 10)
        await asyncio.sleep(0)  # switches task every one iteration.

loop = asyncio.get_event_loop()
futures = [
    asyncio.ensure_future(coro1()),
    asyncio.ensure_future(coro2())
]
loop.run_until_complete(asyncio.gather(*futures))
loop.close()

出局:

1
10
2
20
3
30
4
40
5
50
© www.soinside.com 2019 - 2024. All rights reserved.