FastAPI+Uvicorn单Worker能否处理多并发连接？

问题还原 #

示例代码 #

import asyncio
import datetime
import time

import uvicorn
from fastapi import FastAPI
from starlette.responses import PlainTextResponse

app = FastAPI()


@app.get(path="/sync")
def get_sync():
    print(f"sync: {datetime.datetime.now()}: Before sleep")
    time.sleep(5)
    print(f"sync: {datetime.datetime.now()}: After sleep")
    return PlainTextResponse(content=f"sync: {datetime.datetime.now()}: Hello, World!")


@app.get(path="/async")
async def get_async():
    print(f"async: {datetime.datetime.now()}: Before sleep")
    await asyncio.sleep(5)
    print(f"async: {datetime.datetime.now()}: After sleep")
    return PlainTextResponse(content=f"async: {datetime.datetime.now()}: Hello, World!")


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

操作与问题 #

• 选择GET /sync或GET /async端点
• 从两个浏览器标签页（或cURL）发起并行请求
• 第一个请求会阻塞第二个请求，与预期不符：预期GET /sync在线程池运行、GET /async利用异步IO实现并发，但实际无法做到单Worker下并发。

环境：Python 3.7（64位/Win10），FastAPI与Uvicorn为最新版本，无法使用多Worker。

原因分析 #

1. GET /async端点的阻塞假象：多数浏览器会对同一URL的GET请求做串行化处理（避免重复请求或缓存机制），导致看起来像是阻塞。实际用无缓存的并发工具测试，/async本身是支持并发的——await asyncio.sleep(5)会释放事件循环，让其他请求得以处理。
2. GET /sync端点的真实阻塞：Uvicorn默认单Worker单线程，同步函数会被放到大小为1的线程池执行，第一个请求占用线程后，第二个请求必须等待线程释放，导致阻塞。

解决方法 #

1. 修正GET /async的测试方式 #

不要用同一浏览器的多个标签页测试同一GET端点，改用以下方式：

• 打开多个终端，分别执行：curl http://localhost:1911/async
• 使用Postman的“集合运行器”发起并发请求
• 编写Python异步测试脚本：

import asyncio
import aiohttp

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    tasks = [fetch("http://localhost:1911/async") for _ in range(2)]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

2. 让GET /sync支持并发 #

有两种可行方案：

方案A：启动Uvicorn时指定多线程

修改启动代码，增加threads参数扩大线程池容量：

if __name__ == "__main__":
    uvicorn.run(app=app, host="0.0.0.0", port=1911, threads=4)

同步函数会被分配到不同线程执行，单Worker下也能处理多个并发请求。

方案B：将同步函数改为异步，手动调度阻塞操作

利用asyncio的线程池执行器，将阻塞操作放到线程中运行，释放事件循环处理其他请求（适配Python 3.7）：

import asyncio
import datetime
import time

import uvicorn
from fastapi import FastAPI
from starlette.responses import PlainTextResponse

app = FastAPI()


@app.get(path="/sync")
async def get_sync():
    print(f"sync: {datetime.datetime.now()}: Before sleep")
    loop = asyncio.get_event_loop()
    # 将阻塞操作提交到线程池执行
    await loop.run_in_executor(None, time.sleep, 5)
    print(f"sync: {datetime.datetime.now()}: After sleep")
    return PlainTextResponse(content=f"sync: {datetime.datetime.now()}: Hello, World!")


@app.get(path="/async")
async def get_async():
    print(f"async: {datetime.datetime.now()}: Before sleep")
    await asyncio.sleep(5)
    print(f"async: {datetime.datetime.now()}: After sleep")
    return PlainTextResponse(content=f"async: {datetime.datetime.now()}: Hello, World!")


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

也可以使用FastAPI提供的run_in_threadpool工具函数简化代码：

from fastapi.concurrency import run_in_threadpool

@app.get(path="/sync")
async def get_sync():
    print(f"sync: {datetime.datetime.now()}: Before sleep")
    await run_in_threadpool(time.sleep, 5)
    print(f"sync: {datetime.datetime.now()}: After sleep")
    return PlainTextResponse(content=f"sync: {datetime.datetime.now()}: Hello, World!")

验证 #

修改后，用并发工具测试两个端点，会看到请求同时开始，5秒后同时返回，实现单Worker下的并发处理。

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