如何通过Ctrl+C优雅终止FastAPI及后台线程？

问题根源 #

你当前的代码中，后台线程用while True做无限循环，没有任何退出触发条件；而且Python线程默认是非守护线程，主进程收到Ctrl+C触发KeyboardInterrupt时，这些线程会继续运行，要么阻止进程正常退出，要么导致异常未被正确处理，最终无法优雅终止服务和线程。

解决思路与实现 #

要实现Ctrl+C时正常终止所有线程和FastAPI服务，需要做这几件事：

• 给线程加线程安全的退出标志，让线程能主动退出循环
• 用可中断的等待替代time.sleep，让线程能立即响应退出信号
• 捕获终止信号，触发退出流程
• 在服务关闭时等待所有线程结束，清理资源

修改后的完整代码 #

from fastapi import FastAPI
import threading
import time
import signal
from datetime import datetime, timedelta
import sys
import asyncio

from model import core
from model.database import engine, SessionLocal, get_db
from model.core import Timestamp
from routers.timestamps import router as timestamps_router

core.Base.metadata.create_all(bind=engine)

app = FastAPI()

app.include_router(router=timestamps_router)

# 线程安全的退出事件，用来通知所有后台线程停止
exit_event = threading.Event()
# 保存后台线程引用，方便关闭时等待
background_threads = []

def record_timestamp():
    while not exit_event.is_set():
        db = None
        try:
            db = SessionLocal()
            new_timestamp = Timestamp(timestamp=datetime.utcnow())
            db.add(new_timestamp)
            db.commit()
        except Exception as e:
            if db:
                db.rollback()
            print(f"记录时间戳出错: {e}", file=sys.stderr)
        finally:
            if db:
                db.close()
        # 用wait替代sleep，收到退出信号时能立即唤醒
        exit_event.wait(5)


def delete_old_timestamps():
    while not exit_event.is_set():
        db = None
        try:
            db = SessionLocal()
            one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
            deleted_count = db.query(Timestamp).filter(Timestamp.timestamp < one_minute_ago).delete()
            db.commit()
            print(f"清理旧时间戳: {deleted_count} 条")
        except Exception as e:
            if db:
                db.rollback()
            print(f"清理旧时间戳出错: {e}", file=sys.stderr)
        finally:
            if db:
                db.close()
        exit_event.wait(10)


@app.on_event("startup")
def startup_event():
    t1 = threading.Thread(target=record_timestamp)
    t2 = threading.Thread(target=delete_old_timestamps)
    t1.start()
    t2.start()
    background_threads.extend([t1, t2])


@app.on_event("shutdown")
def shutdown_event():
    exit_event.set()
    # 等待每个线程结束，设置超时避免无限等待
    for thread in background_threads:
        thread.join(timeout=5)
    print("所有后台线程已终止，服务关闭完成")


def handle_termination_signal(signum, frame):
    print("\n收到终止信号，开始优雅关闭...")
    exit_event.set()
    # 触发FastAPI的shutdown流程
    loop = asyncio.get_event_loop()
    loop.create_task(app.shutdown())


# 注册Ctrl+C(SIGINT)和系统终止(SIGTERM)信号处理
signal.signal(signal.SIGINT, handle_termination_signal)
signal.signal(signal.SIGTERM, handle_termination_signal)


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

关键改动说明 #

1. exit_event退出标志：用threading.Event实现线程安全的信号传递，线程循环时检查is_set()，一旦标志被设置就退出循环。
2. exit_event.wait()替代time.sleep：wait()可以被事件提前唤醒，这样收到终止信号时，线程不需要等sleep结束就能立即响应退出。
3. 数据库连接的异常处理：添加try-except-finally块，确保无论数据库操作成功与否，连接都会被关闭，避免资源泄漏。
4. 信号处理函数：捕获SIGINT（Ctrl+C）和SIGTERM（系统终止信号），设置退出标志并触发FastAPI的shutdown事件。
5. shutdown事件等待线程：在服务关闭时，等待所有后台线程结束，确保所有资源都被正确清理。

验证方法 #

运行修改后的代码，按下Ctrl+C，控制台会输出终止提示，等待几秒后所有线程正常结束，进程平稳退出，不会再出现未处理的KeyboardInterrupt异常。

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