如何让Python程序在运行FastAPI服务的同时执行后台任务？

问题根源 #

1. 用uvicorn main:app --reload启动时，if __name__ == "__main__"代码块不会执行——uvicorn是直接导入main模块加载app对象，而非直接运行脚本，导致后台数据采集循环未启动。
2. 分开启动两个进程时，MarketScanner是不同进程的独立实例，内存空间隔离，FastAPI进程无法读取另一个进程的数据。

解决方案：用后台线程共享内存 #

通过FastAPI的启动事件启动后台线程，线程与主服务共享内存空间，既保证数据采集循环运行，又能让接口访问到实时数据。

修改后的完整代码 #

import threading
import time
from fastapi import FastAPI

# 假设你的MarketScanner类定义在此或已导入
class MarketScanner:
    def __init__(self, name, sample_size):
        self.name = name
        self.sample_size = sample_size
        self.market_data = []
        # 加锁保证线程安全，避免数据竞争
        self.lock = threading.Lock()

    def start(self):
        # 初始化逻辑，比如连接数据源
        print(f"MarketScanner {self.name} started")

    def get_new_market_data(self):
        # 模拟获取金融数据的逻辑
        with self.lock:
            self.market_data.append(f"data_{len(self.market_data)+1}")

    def get_market_data_len(self):
        with self.lock:
            return len(self.market_data)

app = FastAPI()
ms1 = MarketScanner(name="ms1", sample_size=10)

# 封装后台数据采集循环
def run_market_scanner(ms):
    ms.start()
    while True:
        ms.get_new_market_data()
        print(f"market data length: {ms.get_market_data_len()}")
        time.sleep(3)

# FastAPI启动时触发，启动后台线程
@app.on_event("startup")
def startup_event():
    # daemon=True：主进程退出时自动终止线程
    scanner_thread = threading.Thread(target=run_market_scanner, args=(ms1,), daemon=True)
    scanner_thread.start()

@app.get("/")
def read_root():
    return {"Welcome": "to analyst."}

@app.get("/mdata/")
def read_market_data_len():
    mdata = ms1.get_market_data_len()
    return {"Market data length": mdata}

关键注意事项 #

• 线程安全：MarketScanner中对共享数据（如market_data）的读写操作必须加锁，避免多线程同时操作导致的数据异常。
• 守护线程：设置daemon=True确保后台线程随FastAPI服务进程退出而自动结束，无需手动清理。

启动方式 #

直接运行uvicorn命令即可，后台循环会自动启动：

uvicorn main:app --reload

此时访问/mdata/接口，就能看到实时增长的市场数据长度。

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