如何实现FastAPI接口启动目录监控Python脚本及相关方案咨询

1. RQ、Celery是否适配需求 #

可以实现，但非必要。
RQ、Celery本质是异步任务调度框架，默认针对一次性短耗时任务设计，要运行常驻的监控任务需要额外关闭超时回收机制，还要维护任务ID与监控目录的映射关系，仅当你需要分布式多机器部署监控实例时才推荐使用，单机器场景完全没必要引入额外的中间件依赖。

2. 现有线程方案的缺陷 #

你当前的多线程方案仅适合测试/极小流量场景，生产环境使用存在明显问题：

• 若FastAPI采用多worker部署（如Uvicorn开多个进程），线程仅归属单个worker进程，后续做终止接口时，请求如果打到其他worker上就找不到对应线程，无法实现终止操作
• 线程无隔离性，某个监控线程崩溃可能牵连整个API服务不可用
• API服务重启后所有监控线程会全部丢失，无法自动恢复

3. 最优实现方案 #

场景1：单机器部署，监控实例数≤100 #

推荐用独立子进程+本地映射存储的轻量方案，比线程稳定、比消息队列简单：

• 每次接口调用时通过subprocess.Popen启动独立的filewatcher脚本进程，将待监控目录作为参数传入
• 用本地KV存储（sqlite或普通json文件即可）维护「监控目录<->进程PID」的映射关系
• 终止接口只要查询对应目录的PID，发送系统信号杀死进程即可，API服务重启后也可以扫描当前运行的监控进程恢复映射表
  示例实现参考：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import subprocess
import json
import os
import signal

app = FastAPI()
# 映射表存储路径，实例数多的话可以换成sqlite
PID_MAP_PATH = "watch_pid_map.json"

class Data(BaseModel):
    watch_dir: str

def load_pid_map():
    if not os.path.exists(PID_MAP_PATH):
        return {}
    with open(PID_MAP_PATH, "r") as f:
        return json.load(f)

def save_pid_map(map_data):
    with open(PID_MAP_PATH, "w") as f:
        json.dump(map_data, f)

@app.post('/register/event')
def register_watchdog(data: Data):
    pid_map = load_pid_map()
    # 避免重复监控同一个目录
    if data.watch_dir in pid_map:
        # 先判断进程是否存活
        try:
            os.kill(pid_map[data.watch_dir], 0)
            return {"status": "success", "msg": "该目录已在监控中"}
        except OSError:
            # 进程已异常退出，删除旧记录
            del pid_map[data.watch_dir]
    # 启动独立子进程运行监控脚本
    proc = subprocess.Popen(
        ["python", "filewatcher.py", data.watch_dir],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True
    )
    pid_map[data.watch_dir] = proc.pid
    save_pid_map(pid_map)
    return {"status": "success"}

@app.post('/stop/event')
def stop_watchdog(data: Data):
    pid_map = load_pid_map()
    if data.watch_dir not in pid_map:
        raise HTTPException(status_code=404, detail="该目录没有正在运行的监控任务")
    pid = pid_map[data.watch_dir]
    try:
        os.kill(pid, signal.SIGTERM)
        del pid_map[data.watch_dir]
        save_pid_map(pid_map)
        return {"status": "success", "msg": "监控已终止"}
    except OSError:
        del pid_map[data.watch_dir]
        save_pid_map(pid_map)
        raise HTTPException(status_code=404, detail="监控进程已不存在")

场景2：大规模分布式部署，监控实例≥100 #

再考虑用Celery实现，将监控任务配置为常驻无超时的异步任务，用Redis维护任务ID与目录的映射关系，支持跨机器调度管理监控实例。

4. 能否实现为Linux后台服务 #

完全可以，分为两部分配置即可：

1. 首先将FastAPI接口本身做成Systemd服务，实现开机自启、后台常驻，示例配置如下（存为/etc/systemd/system/fastapi-watch.service）：

[Unit]
Description=FastAPI Directory Watch Service
After=network.target

[Service]
User=root
WorkingDirectory=/path/to/your/project/directory
ExecStart=/usr/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1
Restart=always

[Install]
WantedBy=multi-user.target

配置完成后执行systemctl daemon-reload && systemctl enable --now fastapi-watch即可启动服务。
2. 你通过接口启动的filewatcher子进程本身已经是后台运行状态，加了start_new_session=True参数后，即使API服务重启，这些监控进程也不会被连带杀死；如果需要更高的可靠性，也可以给监控脚本单独做Systemd服务模板，通过接口动态启停对应实例。

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