FastAPI异步路由使用无限循环引发死锁问题求助

当用单Worker启动Uvicorn时，异步路由的代码直接在事件循环线程中执行。你的route_async里的无限循环没有任何await异步等待操作，会持续占用事件循环，导致整个事件循环被完全阻塞——所有后续请求（包括/hello）都无法被处理，直接造成应用死锁。

而同步路由route_normal不影响其他请求的原因是：FastAPI会把同步函数放到独立的线程池中执行，不会占用事件循环线程，事件循环仍能正常处理其他请求。

方案1：在异步循环中主动让出控制权 #

在异步无限循环里加入await asyncio.sleep(0)（或实际业务需要的异步等待操作），让事件循环有机会处理其他任务。修改后的代码示例：

from fastapi import FastAPI
import random
import asyncio

app = FastAPI()

@app.get("/hello")
async def hello():
    return {"Hello": "World"}

@app.get("/async-fixed")
async def route_async():
    while True:
        print({"route_async": random.randint(0, 10)})
        # 让出事件循环控制权，允许处理其他请求
        await asyncio.sleep(0.001)  # 可根据需求调整间隔

asyncio.sleep(0)会立即把控制权交还给事件循环，处理完其他待执行任务后再回到当前循环。如果不需要高频打印，也可以设置更长的休眠时间（如await asyncio.sleep(1)）。

方案2：用后台任务运行无限循环 #

如果你的无限循环是长期运行的服务逻辑（如监控、定时任务），不应放在路由中触发，而是用FastAPI的后台任务机制或单独线程/进程启动：

方式A：使用BackgroundTasks触发 #

from fastapi import FastAPI, BackgroundTasks
import random
import asyncio

app = FastAPI()

async def long_running_task():
    while True:
        print({"background_task": random.randint(0, 10)})
        await asyncio.sleep(1)

@app.get("/start-task")
async def start_task(background_tasks: BackgroundTasks):
    background_tasks.add_task(long_running_task)
    return {"status": "Task started in background"}

@app.get("/hello")
async def hello():
    return {"Hello": "World"}

注意：这种后台任务会随Worker进程生命周期终止而结束，应用重启后任务会重新启动。若需持久化后台任务，建议使用专门的任务队列（如Celery）或独立服务。

方式B：应用启动时自动运行后台任务 #

如果服务启动后需要持续运行该循环，可通过lifespan钩子在应用启动时创建异步任务：

from fastapi import FastAPI
import random
import asyncio
from contextlib import asynccontextmanager

async def long_running_task():
    while True:
        print({"background_task": random.randint(0, 10)})
        await asyncio.sleep(1)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动后台任务
    task = asyncio.create_task(long_running_task())
    yield
    # 应用关闭时取消任务
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Background task cancelled")

app = FastAPI(lifespan=lifespan)

@app.get("/hello")
async def hello():
    return {"Hello": "World"}

方案3：将同步任务放入线程/进程池 #

如果你的逻辑是CPU密集型（而非IO密集型），异步优势不大，可直接用同步函数放到线程/进程池执行，避免阻塞事件循环：

from fastapi import FastAPI
import random
from concurrent.futures import ThreadPoolExecutor
import time

app = FastAPI()
executor = ThreadPoolExecutor(max_workers=1)

def long_running_sync_task():
    while True:
        print({"sync_task": random.randint(0, 10)})
        time.sleep(1)

@app.get("/start-sync-task")
async def start_sync_task():
    executor.submit(long_running_sync_task)
    return {"status": "Sync task started in thread pool"}

@app.get("/hello")
async def hello():
    return {"Hello": "World"}

• 异步代码的核心是不要阻塞事件循环：任何长时间运行的同步操作或无await的无限循环，都会导致事件循环阻塞，影响整个应用。
• 异步适合IO密集型场景（如数据库查询、HTTP请求），CPU密集型任务更适合用线程/进程池处理。
• 即使使用多Worker模式（如uvicorn ... --workers 4），单个Worker阻塞只会影响该Worker的请求，但依然不推荐在路由中编写无限循环。

内容的提问来源于stack exchange，提问作者tombreplacer