FastAPI+Uvicorn异步接口阻塞主线程及请求取消失效问题

在uvicorn workers=1的情况下，async def接口中的耗时操作会阻塞主线程——例如调用/uploadPdf接口运行期间，服务器无法处理新请求（比如无法调用/firstTest）。此外，取消/uploadPdf请求后，文本提取任务仍会持续执行，最终导致Uvicorn进程因客户端断开连接而崩溃。

可复现代码 #

import asyncio
from fastapi import FastAPI, Request, UploadFile, File
import uvicorn
import time
import io
import PyPDF2
from api.decorators import cancel_on_disconnect, threaded

app = FastAPI()

@app.get("/firstTest")
async def hello(request: Request):
    while not await request.is_disconnected():
        print("I am still alive")
        await asyncio.sleep(1)

    print("done")
    return "Hello, world"

# 尝试并行处理的耗时函数
@threaded
def processPages(pages):
    text = ''
    for i in range(len(pages)):
        pageObj = pages[i]
        text = text + pageObj.extract_text() + ' '
    return text

@app.post('/uploadPdf')
async def uploadPdf(request: Request, file: UploadFile = File(...)):
    print("uploading file")
    bytes = await file.read()
    doc = io.BytesIO(bytes)
    pdfReader = PyPDF2.PdfReader(doc)
    t4 = time.time()
    t = processPages(pdfReader.pages)
    text = t.result_queue.get()  # 同步阻塞导致事件循环卡住
    print(f'it took {time.time() - t4}s to extract text')
    return text

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

已尝试的解决方法 #

• 使用threading.Thread，调用thread.start()后执行thread.join()：同步join直接阻塞事件循环，无法解决问题
• 使用@background装饰器基于run_in_executor：

  def background(f):
      def wrapped(*args, **kwargs):
          return asyncio.get_event_loop().run_in_executor(None, f, *args, **kwargs)
  
      return wrapped
  

• 使用基于Queue的@threaded装饰器：

  def threaded(f, daemon=False):
      import queue
      def wrapped_f(q: queue.Queue, *args, **kwargs):
          '''调用被装饰函数并将结果放入队列'''
          ret = f(*args, **kwargs)
          q.put(ret, block=False)
      def wrap(*args, **kwargs):
          '''装饰器返回的函数，启动新线程执行wrapped_f并返回带结果队列的线程对象'''
          q = queue.Queue()
  
          t = threading.Thread(target=wrapped_f, args=(q,)+args, kwargs=kwargs)
          t.daemon = daemon
          t.start()
          t.result_queue = q        
          return t
      return wrap
  

解决方案 #

核心问题分析 #

1. 主线程阻塞：原代码中t.result_queue.get()是同步阻塞调用，会卡住asyncio事件循环，导致主线程无法处理其他请求。
2. 任务无法取消：后台线程未监听请求取消信号，即使客户端断开连接仍会继续执行；同时未处理asyncio.CancelledError，引发进程崩溃。

修改后的代码实现 #

import asyncio
from fastapi import FastAPI, Request, UploadFile, File
import uvicorn
import time
import io
import PyPDF2
import threading

app = FastAPI()

@app.get("/firstTest")
async def hello(request: Request):
    while not await request.is_disconnected():
        print("I am still alive")
        await asyncio.sleep(1)

    print("done")
    return "Hello, world"

# 改造耗时函数，支持取消信号
def processPages(pages, stop_event: threading.Event):
    text = ''
    total_pages = len(pages)
    for i in range(total_pages):
        # 检查是否需要停止任务
        if stop_event.is_set():
            print("任务被取消，提前退出")
            return ""
        pageObj = pages[i]
        text += pageObj.extract_text() + ' '
    return text

@app.post('/uploadPdf')
async def uploadPdf(request: Request, file: UploadFile = File(...)):
    print("uploading file")
    bytes = await file.read()
    doc = io.BytesIO(bytes)
    pdfReader = PyPDF2.PdfReader(doc)
    t4 = time.time()
    
    # 创建停止事件，用于通知线程取消任务
    stop_event = threading.Event()
    
    try:
        # 使用asyncio.to_thread运行耗时函数（Python3.9+），异步获取结果不阻塞事件循环
        text = await asyncio.to_thread(processPages, pdfReader.pages, stop_event)
        print(f'it took {time.time() - t4}s to extract text')
        return text
    except asyncio.CancelledError:
        # 捕获请求取消异常，触发线程停止信号
        stop_event.set()
        print("请求已取消，终止文本提取任务")
        raise  # 重新抛出异常，让FastAPI正确处理请求取消
    finally:
        pass

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

关键改进点 #

1. 避免同步阻塞：用asyncio.to_thread（或loop.run_in_executor）将耗时函数放入线程池执行，通过await异步获取结果，事件循环可正常处理其他请求。
2. 支持任务取消：通过threading.Event传递取消信号，耗时函数循环中检查事件状态；捕获请求取消异常后触发停止信号，让线程提前退出。
3. 简化线程管理：利用Python标准库线程池替代自定义装饰器，降低代码复杂度。

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