FastAPI客户端断连后仍处理请求，如何终止或即时响应？

我有一个运行在Gunicorn Web服务器上的FastAPI应用，网关接口为ASGI。我在应用中模拟了耗时较长的响应场景，原本期望当客户端断开请求连接（如关闭浏览器标签页）时，ASGI能够中止或终止Gunicorn worker中的任务，但实际发现FastAPI仍会完成整个处理流程。

相关日志：

api          | TRACE:    HTTP connection made
api          | TRACE:    ASGI [4] Started scope={'type': 'http', 'asgi': {'version': '3.0', 'spec_version': '2.3'}, 'http_version': '1.0', 'server': None, 'client': None, 'scheme': 'http', 'root_path': '', 'headers': '<...>', 'method': 'GET', 'path': '/long-time', 'raw_path': b'/long-time', 'query_string': b''}
api-proxy    | 172.23.0.1 - - [12/Dec/2022:07:31:46 +0000] "GET /long-time HTTP/1.1" 499 "e9224322f5f65f94a54e4a6ca812ae72" generate 0 bytes in 3.051 "http://localhost:8000/docs" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36" "-
api          | TRACE:    HTTP connection lost
api          | [2022-12-12 15:31:53,295] [INFO] [/router.py:810]: Ready to return
api          | TRACE:    ASGI [4] Receive {'type': 'http.disconnect'}
api          | TRACE:    ASGI [4] Send {'type': 'http.response.start', 'status': 200, 'headers': '<...>'}
api          | TRACE:    ASGI [4] Send {'type': 'http.response.body', 'body': '<4 bytes>'}
api          | TRACE:    ASGI [4] Completed

示例代码：

async def my_func_1():
    """
    my func 1
    """
    await asyncio.sleep(10)

    return "zzzzzzzz"


@api_router.get("long-time")
async def root():
    """
    my home route
    """
    a = await asyncio.gather(my_func_1())

    fastapi_logger.info("Ready to return")
    
    return a

请问是否存在可配置或可编程的方式来终止请求处理，或即时返回响应而非等待API端点函数执行完毕？

1. 先确保使用支持ASGI的Worker进程 #

Gunicorn默认的同步Worker无法正确传递ASGI的连接断开事件，必须切换为Uvicorn的ASGI Worker启动服务：

gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app

2. 手动监听连接状态并终止任务 #

FastAPI允许在端点中注入Request对象，通过request.is_disconnected()方法实时检查客户端连接状态。你需要将长耗时任务拆分为短步骤，定期检查连接，或者将任务包装为异步任务并主动取消。

修改后的示例代码： #

from fastapi import Request, APIRouter
import asyncio
import logging

fastapi_logger = logging.getLogger(__name__)
api_router = APIRouter()

async def my_func_1(request: Request):
    # 拆分长休眠为多个1秒步骤，每次休眠前检查连接
    for _ in range(10):
        if await request.is_disconnected():
            raise asyncio.CancelledError
        await asyncio.sleep(1)
    return "zzzzzzzz"

@api_router.get("/long-time")
async def root(request: Request):
    try:
        task = asyncio.create_task(my_func_1(request))
        result = await task
        fastapi_logger.info("Ready to return")
        return result
    except asyncio.CancelledError:
        fastapi_logger.info("Client disconnected, task cancelled")
        return {"detail": "Request cancelled due to client disconnection"}

3. 针对不可拆分的任务：结合连接检查协程 #

如果耗时任务无法拆分，可以同时启动一个专门检查连接状态的协程，一旦检测到断开就终止所有任务：

async def check_connection(request: Request):
    while True:
        if await request.is_disconnected():
            raise asyncio.CancelledError
        await asyncio.sleep(0.5)

async def my_func_1():
    await asyncio.sleep(10)
    return "zzzzzzzz"

@api_router.get("/long-time")
async def root(request: Request):
    try:
        result, _ = await asyncio.gather(
            my_func_1(),
            check_connection(request),
            return_exceptions=False
        )
        fastapi_logger.info("Ready to return")
        return result
    except asyncio.CancelledError:
        fastapi_logger.info("Client disconnected, task cancelled")
        return {"detail": "Client disconnected"}

关键说明 #

• ASGI协议不会自动终止正在执行的异步任务，必须由开发者主动处理连接断开事件。
• 日志中的http.disconnect事件通常在任务执行完毕后才会被接收，因此无法依赖该事件实时终止任务，必须主动调用request.is_disconnected()检查。
• 如果是CPU密集型同步任务，需要用asyncio.to_thread将其放入线程池，通过共享状态标记让任务自行终止（线程无法直接取消）。

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