网站软件免费下载wordpress 插件漏洞

目录

1. 问题背景

2. 原因分析

3. 常见解决方案

3.1 方案一：单 worker 模式（简单粗暴）

3.2 方案二：文件锁（单机可用，跨平台推荐）

示例代码（使用 filelock，兼容 Linux 和 Windows）

3.3 方案三：使用 Redis/数据库 JobStore（推荐）

示例代码（Redis JobStore）

3.4 方案四：独立 Scheduler 进程（解耦推荐）

示例

4. 方案对比

5. 总结与推荐

---

1. 问题背景

在 FastAPI 中集成 APScheduler，可以方便地执行定时任务。但当我们用 Uvicorn 启动应用时，如果指定了 --workers > 1，就会出现 每个 worker 进程都会初始化一份 APScheduler 调度器 的情况。

例如：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

结果是同一个任务被 重复执行 4 次（每个 worker 都跑一遍），这在生产环境是无法接受的。

---

2. 原因分析

• Uvicorn 的多进程模式通过 multiprocessing fork 出多个完全独立的 Python 进程。

• 每个进程都会加载一份 FastAPI 应用实例，执行 @app.on_event("startup") 钩子。

• 如果在 startup 里初始化 APScheduler，就会导致 每个进程都启动了自己的调度器。

换句话说：Uvicorn 没有 master/worker 角色区分，每个 worker 都是“主进程”。

---

3. 常见解决方案

3.1 方案一：单 worker 模式（简单粗暴）

只启动一个进程，避免重复加载：

uvicorn main:app --workers 1

适合任务量不大、并发不高的应用。 但缺点是失去了多进程并发能力。

---

3.2 方案二：文件锁（单机可用，跨平台推荐）

在应用启动时尝试获取文件锁，只有第一个获取锁的进程会启动 APScheduler，其他进程跳过。

示例代码（使用 filelock，兼容 Linux 和 Windows）

import os
import logging
from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from filelock import FileLock, Timeout
​
app = FastAPI()
LOCK_FILE = "scheduler.lock"
​
def my_job():print(f"[PID={os.getpid()}] running scheduled task...")
​
@app.on_event("startup")
async def startup_event():lock = FileLock(LOCK_FILE)try:lock.acquire(timeout=0.1)scheduler = AsyncIOScheduler()scheduler.add_job(my_job, "interval", seconds=10, id="unique_job", replace_existing=True)scheduler.start()
​app.state.scheduler = schedulerapp.state.lock = locklogging.info(f"[PID={os.getpid()}] Scheduler started (lock acquired).")
​except Timeout:logging.info(f"[PID={os.getpid()}] Another process holds the scheduler lock. Skipping.")
​
@app.on_event("shutdown")
async def shutdown_event():if hasattr(app.state, "scheduler"):app.state.scheduler.shutdown()if hasattr(app.state, "lock"):app.state.lock.release()logging.info(f"[PID={os.getpid()}] Scheduler stopped and lock released.")

🔎 特点：

• 适合 单机部署；

• 锁文件路径必须可写；

• 跨平台可用（Linux/Windows 都支持）。

---

3.3 方案三：使用 Redis/数据库 JobStore（推荐）

APScheduler 支持 Redis、MongoDB、SQLAlchemy 等作为 JobStore。 即使多个进程启动调度器，它们共享相同的任务存储，不会重复执行。

示例代码（Redis JobStore）

from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.redis import RedisJobStore
​
app = FastAPI()
​
def my_job():print("Running job...")
​
@app.on_event("startup")
async def startup_event():jobstores = {"default": RedisJobStore(host="localhost", port=6379, db=0)}scheduler = AsyncIOScheduler(jobstores=jobstores)scheduler.add_job(my_job, "interval", seconds=10, id="unique_job", replace_existing=True)scheduler.start()app.state.scheduler = scheduler
​
@app.on_event("shutdown")
async def shutdown_event():if hasattr(app.state, "scheduler"):app.state.scheduler.shutdown()

🔎 特点：

• 适合 多进程、多机集群 部署；

• 保证任务全局唯一；

• 需要 Redis 或数据库支持。

---

3.4 方案四：独立 Scheduler 进程（解耦推荐）

将 APScheduler 独立成单独的进程或服务，只负责任务调度；FastAPI 应用只处理 API 请求。

示例

# scheduler.py
import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler
​
def my_job():print("Task executed")
​
async def main():scheduler = AsyncIOScheduler()scheduler.add_job(my_job, "interval", seconds=10)scheduler.start()await asyncio.Event().wait()
​
if __name__ == "__main__":asyncio.run(main())

运行：

python scheduler.py
uvicorn main:app --workers 4

🔎 特点：

• 彻底避免重复加载；

• 解耦，适合中大型项目。

---

4. 方案对比

方案	优点	缺点
单 worker	简单稳定	无法利用多进程并发
文件锁	跨平台，代码简单，单机有效	多机部署无效
Redis/DB JobStore	多进程/多机可用，全局唯一	依赖额外存储
独立 scheduler 进程	彻底解耦，灵活扩展	需要额外部署进程

---

5. 总结与推荐

• 开发环境 / 单机部署：可以用 文件锁 简单解决。

• 生产环境 / 多机部署：推荐用 Redis/DB JobStore 或者 独立 scheduler 进程。

• 任务量大/耗时任务：建议用 APScheduler + Celery/RQ，调度和执行分离。

---

最佳实践：

FastAPI（Uvicorn 多 worker）只处理请求，定时任务交给单独的 scheduler 服务，或者 Redis JobStore 管理。 这样既能保证任务不会重复执行，又能发挥多进程并发的优势。

---