python - 如何使用 uvicorn 创建的事件循环？

Question

我正在使用 uvicorn，我需要使用现有的事件循环。我正在使用以下命令:

loop = asyncio.get_event_loop()

但是当我使用这一行时，代码卡住了。但是如果我使用 new_event_loop，我会得到以下错误:

RuntimeError: Task <Task pending coro=<open_connection() running at /usr/lib/python3.6/asyncio/streams.py:81> cb=[_release_waiter(<Future pendi...sk._wakeup()]>)() at /usr/lib/python3.6/asyncio/tasks.py:316]> got Future <Future pending> attached to a different loop

请帮助我了解如何使用 uvicorn 创建的现有循环。

Answer 1

你可以阅读这个问题@https://github.com/encode/uvicorn/issues/706

基本上，您必须创建自己的事件循环并将其传递给 uvicorn。

import asyncio
from uvicorn import Config, Server


async def app(scope, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            [b'content-type', b'text/plain'],
        ]
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello, world!',
    })


loop = asyncio.new_event_loop()

config = Config(app=app, loop=loop)
server = Server(config)
loop.run_until_complete(server.serve())