FastAPI异步使用疑问：为何未异步处理请求及阻塞函数相关问题

我编写了以下server.py和client.py代码：

server.py #

from fastapi import FastAPI
import uvicorn
import time

app = FastAPI()

@app.post('/upload')
async def upload(file: dict):
    print('hello')

    time.sleep(1)
    
    return file

if __name__ == '__main__':
    uvicorn.run(app, host='', port=1024)

client.py #

import requests
import threading

def thread_main(a):
    response = requests.post(url, json={'0':0, '1':1, '2':2})
    if response.status_code == 200:
        print(a, response.json())
    else:
        print(a, 'wrong')

url = 'http://IP:1024/upload'
for i in range(10):
    thread = threading.Thread(target=thread_main, args=(i,))
    thread.start()

当前服务器日志输出为：

hello
INFO:     IP:port - "POST /upload HTTP/1.1" 200 OK
hello
INFO:     IP:port - "POST /upload HTTP/1.1" 200 OK
hello
INFO:     IP:port - "POST /upload HTTP/1.1" 200 OK
...

但我期望的日志输出是：

hello
hello
hello
...
INFO:     IP:port - "POST /upload HTTP/1.1" 200 OK
INFO:     IP:port - "POST /upload HTTP/1.1" 200 OK
INFO:     IP:port - "POST /upload HTTP/1.1" 200 OK
...

想问：

1. FastAPI中能否实现该效果？我原以为用async定义函数就能让服务器异步处理请求，但实际并未生效。
2. 如何判断一个函数是否为阻塞函数？
3. 如何寻找阻塞函数对应的异步替代函数？

1. 为什么当前代码没实现异步？ #

你用async定义了接口函数，但里面调用了阻塞的同步函数time.sleep(1)——这个函数会直接卡住FastAPI的事件循环，导致后续请求必须等当前请求的sleep完成才能被处理，所以日志会是一个hello跟着一条响应日志，没法并行处理请求。

2. 实现期望效果的方法 #

要让FastAPI真正异步处理请求，需要把阻塞的同步操作替换成异步版本：

• 将time.sleep(1)换成异步的asyncio.sleep(1)，asyncio.sleep不会阻塞事件循环，会把控制权交还给事件循环，让它可以同时处理其他请求。

修改后的server.py代码：

from fastapi import FastAPI
import uvicorn
import asyncio  # 导入asyncio库

app = FastAPI()

@app.post('/upload')
async def upload(file: dict):
    print('hello')

    await asyncio.sleep(1)  # 替换为异步sleep
    
    return file

if __name__ == '__main__':
    uvicorn.run(app, host='', port=1024)

修改后，服务器会先一次性打印所有10个hello，等待1秒后再批量输出响应日志，完全符合你的期望。

3. 判断阻塞函数的方法 #

判断一个函数是否阻塞，看这几点：

• 函数是同步定义（无async关键字），且执行时会占用CPU或等待IO（比如文件读写、网络请求、sleep），期间不会释放CPU给其他任务。
• 典型的阻塞函数：time.sleep()、requests.get()、open().read()等，它们执行时会让当前线程暂停，直到操作完成。
• 注意：异步函数（带async）如果内部调用了阻塞同步函数，整个异步函数也会变成阻塞的。

4. 寻找异步替代函数的方法 #

• 优先找官方或主流库提供的异步版本：
  • 同步sleep → asyncio.sleep()（Python标准库）
  • 同步HTTP请求 → aiohttp（第三方异步HTTP库）
  • 同步文件读写 → aiofiles（第三方异步文件操作库）
• 如果找不到现成的异步库，可以把阻塞函数放到线程池执行，用asyncio.to_thread()（Python 3.9+）或loop.run_in_executor()包装，避免阻塞事件循环。示例：

  import asyncio
  import time
  
  async def upload(file: dict):
      print('hello')
      # 将阻塞的time.sleep放到线程池执行
      await asyncio.to_thread(time.sleep, 1)
      return file
  

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