在 FastAPI 应用中，如果需要在异步路由中执行无限循环，直接使用 while True 可能会导致整个应用死锁，其他路由无法响应。这是因为异步函数在执行时，如果没有适当的让出控制权，会阻塞事件循环，导致 FastAPI 无法处理其他请求。下面将介绍两种避免死锁的解决方案。
使用 BackgroundTasks
FastAPI 提供了 BackgroundTasks 类，可以将耗时任务放入后台执行，从而避免阻塞主线程。这种方式非常适合处理无限循环任务。
from fastapi import FastAPI, BackgroundTasks import random app = FastAPI() @app.get("/hello") async def hello(): return {"Hello": "World"} @app.get("/normal") def route_normal(): while True: print({"route_normal": random.randint(0, 10)}) @app.get("/async") async def route_async(background_tasks: BackgroundTasks): def background_task(): while True: print({"route_async": random.randint(0, 10)}) background_tasks.add_task(background_task) return {"message": "Background task started"}
代码解释：
- 导入 BackgroundTasks 类。
- 在异步路由 route_async 中，声明一个 background_tasks 参数，类型为 BackgroundTasks。FastAPI 会自动注入该对象。
- 定义一个 background_task 函数，其中包含无限循环。
- 使用 background_tasks.add_task() 将 background_task 函数添加到后台任务队列。
- 路由函数立即返回一个消息，表示后台任务已启动。
注意事项：
- 后台任务的执行与主线程是并发的，因此需要注意线程安全问题。
- BackgroundTasks 适用于执行不需要立即返回结果的任务。
使用 asyncio.sleep()
另一种解决方案是在无限循环中加入 asyncio.sleep()，让出控制权，允许事件循环处理其他任务。
import asyncio from fastapi import FastAPI import random app = FastAPI() @app.get("/hello") async def hello(): return {"Hello": "World"} @app.get("/normal") def route_normal(): while True: print({"route_normal": random.randint(0, 10)}) @app.get("/async") async def route_async(): while True: await asyncio.sleep(0) # do a sleep here so that the main thread can do its magic, at least once per loop, changing the sleep duration will allow the main thread to process other threads longer, please read up more on the specifics print({"route_async": random.randint(0, 10)})
代码解释：
- 导入 asyncio 模块。
- 在异步路由 route_async 的无限循环中，使用 await asyncio.sleep(0) 让出控制权。
注意事项：
- asyncio.sleep(0) 会立即让出控制权，允许事件循环处理其他任务。
- 可以调整 asyncio.sleep() 的参数，控制让出控制权的时间。参数越大，其他任务获得执行的机会越多，但当前任务的执行速度会降低。
- 这种方法适用于对实时性要求不高的任务。
总结
在 FastAPI 异步路由中使用无限循环时，需要特别注意避免阻塞事件循环。使用 BackgroundTasks 可以将任务放入后台执行，而使用 asyncio.sleep() 可以让出控制权。选择哪种方案取决于具体的应用场景和需求。如果任务不需要立即返回结果，且对实时性要求不高，建议使用 BackgroundTasks。如果需要在循环中进行一些操作，且对实时性有一定的要求，可以使用 asyncio.sleep()。