Asyncio 使用 uvicorn 和 fastapi 同时运行多个异步函数答案

【问题标题】：Asyncio running multiple asynconrous functions at once with uvicorn and fastapiAsyncio 使用 uvicorn 和 fastapi 同时运行多个异步函数
【发布时间】：2022-12-19 19:04:56
【问题描述】：

我在使用 uvicorn (fastapi) 运行我自己的异步函数时遇到了一些问题

我的目标是能够运行 queue manager 函数，以及 uvicorn webapp。但是，我的start 函数似乎阻止了uvicorn.run 代码的运行。


async def queue_manager(self):
        while True:
            job = await self.queue.get()
            print(job)
            await asyncio.sleep(2)

async def start():
    task1 = asyncio.create_task(queue_manager())
    await asyncio.gather(task1)

app = FastAPI()
app.include_router(server.router)

if __name__ == "__main__":
    asyncio.run(start())
    uvicorn.run("main:app", host="0.0.0.0", port=5959)

为了解决这个问题，我尝试在我的start 函数中运行uvicorn.run，但这会导致以下错误：

line 35, in <module>
    asyncio.run(start())
line 17, in start
    await uvicorn.run("main:app", host="0.0.0.0", port=5959)
RuntimeError: asyncio.run() cannot be called from a running event loop

async def start():
    task1 = asyncio.create_task(custom_instance_manager.queue_manager())
    task2 = asyncio.create_task(uvicorn.run(
        "main:app", host="0.0.0.0", port=5959))
    await asyncio.gather(task1, task2)

if __name__ == "__main__":
    asyncio.run(start())

【问题讨论】：

标签： python python-asyncio fastapi uvicorn

【解决方案1】：

您可以使用 startup 事件。

async def queue_manager(self):
        while True:
            job = await self.queue.get()
            print(job)
            await asyncio.sleep(2)


app = FastAPI()
app.include_router(server.router)

@app.on_event('startup')
async def start():
    asyncio.create_task(queue_manager())

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=5959)

【讨论】：

您应该保留该任务的引用，否则 GC 可以清理它。

【解决方案2】：

您可以使用具有快速 api 配置的 uvicorn 服务器异步函数，而不是使用 uvicorn.run

async def run_server():
    api = configure_fastapi()
    config = uvicorn.Config(api, port=8021, host='0.0.0.0')
    server = uvicorn.Server(config)
    await server.serve()

【讨论】：