FastAPI自定义连接关闭处理器问题：原生shutdown机制失效

你的问题核心是自定义信号处理函数覆盖了FastAPI/Starlette原生的信号处理逻辑，导致框架自身的shutdown机制（比如lifespan yield后的代码、注册的shutdown事件）无法触发。解决办法是在自定义处理逻辑中先调用原生的信号处理函数，再执行你的任务取消逻辑。

具体实现步骤 #

1. 保存FastAPI原本的SIGTERM和SIGINT信号处理函数
2. 在自定义shutdown函数中，先调用原生处理函数保证默认流程执行
3. 再执行你的优雅取消任务逻辑

代码示例 #

import signal
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI
import logging

logger = logging.getLogger(__name__)

# 保存原生信号处理函数的全局变量
original_sigterm_handler = None
original_sigint_handler = None

async def async_shutdown_tasks():
    """你的自定义优雅取消任务逻辑：响应挂起请求、终止队列任务等"""
    logger.info("开始处理挂起请求，取消队列任务...")
    # 这里添加你的实际业务代码：
    # 比如关闭任务队列、向所有挂起连接返回响应、取消asyncio任务等

def handle_shutdown(signum, frame):
    # 优先调用FastAPI原生的信号处理函数，保证默认shutdown流程执行
    if signum == signal.SIGTERM and original_sigterm_handler:
        original_sigterm_handler(signum, frame)
    elif signum == signal.SIGINT and original_sigint_handler:
        original_sigint_handler(signum, frame)
    
    # 执行自定义异步shutdown逻辑
    asyncio.run(async_shutdown_tasks())

@asynccontextmanager
async def lifespan(instance: FastAPI):
    global original_sigterm_handler, original_sigint_handler
    # 保存框架原生的信号处理函数
    original_sigterm_handler = signal.getsignal(signal.SIGTERM)
    original_sigint_handler = signal.getsignal(signal.SIGINT)
    
    # 注册自定义信号处理
    signal.signal(signal.SIGTERM, handle_shutdown)
    signal.signal(signal.SIGINT, handle_shutdown)
    
    logger.info("服务启动，信号处理已注册")
    yield 
    # 这里的代码会在原生shutdown流程中执行（如果需要）
    logger.info("原生shutdown流程执行完毕")

关键说明 #

• 当你直接调用signal.signal()注册新处理函数时，会覆盖Starlette初始化时注册的信号处理逻辑，而原生逻辑负责触发lifespan的shutdown阶段、关闭服务器等流程。
• 通过signal.getsignal()提前保存原生处理函数，在自定义逻辑中先调用它，就能同时保留FastAPI的默认shutdown流程和你的自定义任务取消逻辑。
• 如果你的shutdown逻辑是异步的，需要在同步的信号处理函数中用asyncio.run()来执行异步代码（信号处理函数必须是同步的）。

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