FastAPI 如何在 FastAPI 中停止循环

在本文中，我们将介绍如何在 FastAPI 中停止循环。FastAPI 是一个高性能的 Web 框架，基于 Python 的类型提示，具有快速和简单的特点。在一些应用场景中，我们可能需要在应用关闭时停止正在运行的循环，以确保应用的正常关闭。

阅读更多：FastAPI 教程

FastAPI 的启动和关闭事件

在 FastAPI 中，我们可以使用 Startup/Shutdown 事件来执行在应用启动和关闭时需要运行的代码。FastAPI 提供了四个针对 Startup/Shutdown 事件的装饰器：

app.on_event("startup") ：在应用启动时调用的函数可以使用此装饰器进行装饰。
app.on_event("shutdown")：在应用关闭时调用的函数可以使用此装饰器进行装饰。
router.on_event("startup")：在路由启动时调用的函数可以使用此装饰器进行装饰。
router.on_event("shutdown")：在路由关闭时调用的函数可以使用此装饰器进行装饰。

使用 Startup/Shutdown 事件，我们可以在应用启动时执行一些初始化操作并在应用关闭时执行清理操作。

停止循环的实现

要停止 FastAPI 应用中的循环，我们可以使用多线程的方式，将循环代码放在一个独立的线程中，并在应用关闭时通过设置一个标志来停止该线程的运行。下面是一个示例代码：

import threading
from fastapi import FastAPI
from fastapi import BackgroundTasks

app = FastAPI()
stop_flag = threading.Event()

def long_running_task():
    while not stop_flag.is_set():
        # 循环任务逻辑
        pass

def start_long_running_task(background_tasks: BackgroundTasks):
    t = threading.Thread(target=long_running_task)
    t.start()
    background_tasks.add_task(stop_long_running_task)

def stop_long_running_task():
    stop_flag.set()

@app.on_event("startup")
async def startup_event(background_tasks: BackgroundTasks):
    start_long_running_task(background_tasks)

@app.on_event("shutdown")
async def shutdown_event():
    stop_long_running_task()

在上面的代码中，我们使用 threading.Event() 创建了一个 stop_flag 事件对象，用于控制循环线程的运行。long_running_task() 是一个演示用的循环任务，通过检查 stop_flag 的状态来决定是否停止任务的执行。

start_long_running_task() 函数用于在应用启动时启动循环任务的线程，并将停止任务的函数 stop_long_running_task() 添加到后台任务中。stop_long_running_task() 函数用于修改 stop_flag 的状态，以达到停止循环任务的目的。

最后，我们使用 app.on_event("startup") 和 app.on_event("shutdown") 装饰器来分别注册启动事件和关闭事件，以确保在应用启动和关闭时执行相应的操作。

这样，我们就可以在 FastAPI 应用中控制循环任务的启动和停止了。