在 FastAPI 的 @app.on_event("startup") 中启动后台线程

在 FastAPI 项目中，某些任务可能需要在应用启动时运行，并在后台持续执行，如定时清理数据库、监控资源或定时获取数据等。为此，我们可以在 FastAPI 的 @app.on_event("startup") 中启动后台线程，确保应用在启动后立刻开始这些任务。

说明：@app.on_event 已被 FastAPI 标记为弃用。存量项目仍可能见到这种写法，理解它仍有价值；新项目建议优先使用 lifespan，文末给出对应写法。

背景知识

FastAPI 是一个支持异步编程的 Python Web 框架，运行在异步事件循环上，能够高效处理 I/O 密集型任务。不过，在应用启动的过程中，有些同步阻塞任务不适合直接运行在主事件循环里。这时，可以使用 Python 标准库的 threading 模块，结合 @app.on_event("startup") 启动线程，以便在后台执行这些任务。

在 `startup` 中启动后台线程的方式

有几种方式可以在 startup 中启动后台线程，以下是几种常见的方法。

1. 使用 `threading.Thread` 启动后台线程

threading.Thread 是 Python 标准库提供的多线程模块，可以在启动应用时启动一个守护线程，让它在后台运行。

示例代码如下：

import threading
import time
from fastapi import FastAPI

app = FastAPI()

def background_task():
    while True:
        print("Running background task...")
        time.sleep(10)  # 模拟一个需要重复执行的任务

@app.on_event("startup")
async def startup():
    # 创建并启动一个守护线程，以便在应用启动时开始执行后台任务
    thread = threading.Thread(target=background_task, daemon=True)
    thread.start()
    print("Background thread has started.")

@app.on_event("shutdown")
async def shutdown():
    print("Application is shutting down.")

解释：

守护线程：daemon=True 表示该线程是守护线程，应用关闭时不会等待该线程结束，确保应用可以快速关闭。
持续运行：background_task 函数中的 while True 循环使得该线程持续执行，可以通过 time.sleep() 控制任务的频率。

2. 使用 `concurrent.futures.ThreadPoolExecutor` 启动线程池

如果有多个后台任务需要并行运行，concurrent.futures.ThreadPoolExecutor 是更好的选择，它可以创建一个线程池来管理多个后台任务。

from concurrent.futures import ThreadPoolExecutor
import time
from fastapi import FastAPI

app = FastAPI()
executor = ThreadPoolExecutor(max_workers=3)

def task_one():
    while True:
        print("Running task one...")
        time.sleep(10)

def task_two():
    while True:
        print("Running task two...")
        time.sleep(15)

@app.on_event("startup")
async def startup():
    # 使用线程池启动多个后台任务
    executor.submit(task_one)
    executor.submit(task_two)
    print("Background tasks have started.")

@app.on_event("shutdown")
async def shutdown():
    executor.shutdown(wait=False)  # 停止所有后台任务
    print("Application is shutting down.")

解释：

多个任务：每个任务可以独立定义，并通过 executor.submit() 添加到线程池。
线程池：ThreadPoolExecutor 会自动管理线程，最大线程数通过 max_workers 控制。

3. 使用 `asyncio.create_task` 启动异步任务

如果任务是异步的，可以使用 asyncio.create_task 启动一个异步任务，这样就不需要创建新的线程。异步任务适合用于 I/O 密集型操作，比如网络请求、数据库访问等。

import asyncio
from fastapi import FastAPI

app = FastAPI()

async def async_background_task():
    while True:
        print("Running async background task...")
        await asyncio.sleep(10)  # 模拟异步操作

@app.on_event("startup")
async def startup():
    # 使用 asyncio 启动异步任务
    asyncio.create_task(async_background_task())
    print("Async background task has started.")

@app.on_event("shutdown")
async def shutdown():
    print("Application is shutting down.")

解释：

异步任务：async_background_task 使用 async def 声明，是一个异步函数。
事件循环：asyncio.create_task 在主事件循环上创建任务，而不需要额外线程。这对 I/O 密集型任务非常高效，因为异步任务会在等待 I/O 操作时让出控制权，不会阻塞主线程。

注意事项

守护线程：确保后台线程设置为守护线程（daemon=True），否则 FastAPI 关闭时会等待线程结束。
资源清理：在 @app.on_event("shutdown") 中执行清理操作，如释放数据库连接、关闭文件等，防止资源泄露。
选择合适的方式：根据任务类型选择合适的实现方式：
- I/O 密集型任务使用 asyncio.create_task 启动异步任务。
- 同步阻塞型任务可以考虑 threading.Thread 或 ThreadPoolExecutor。
- CPU 密集型任务更适合进程池、独立 worker 或任务队列，避免被 GIL 限制拖慢主进程。
- 多个并发任务使用 ThreadPoolExecutor 来管理。

新项目推荐使用 `lifespan`

FastAPI 新项目更推荐通过 lifespan 管理启动和关闭逻辑。下面示例用 threading.Event 控制后台线程退出，避免只依赖守护线程：

import threading
import time
from contextlib import asynccontextmanager
from fastapi import FastAPI

stop_event = threading.Event()

def background_task():
    while not stop_event.is_set():
        print("Running background task...")
        time.sleep(10)

@asynccontextmanager
async def lifespan(app: FastAPI):
    thread = threading.Thread(target=background_task)
    thread.start()
    try:
        yield
    finally:
        stop_event.set()
        thread.join(timeout=5)

app = FastAPI(lifespan=lifespan)