如何实现FastAPI+APScheduler多Worker部署时任务不重复触发（Redis同步）

依赖安装 #

首先安装所需依赖包：

pip install fastapi uvicorn apscheduler redis

完整代码示例 #

from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.triggers.interval import IntervalTrigger
import redis
import time

app = FastAPI()

# 配置Redis连接参数
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0

# 初始化Redis客户端用于分布式锁
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)

# 定义定时任务
def scheduled_task():
    # 分布式锁key，确保任务唯一锁定
    lock_key = "scheduled_task_lock"
    # 锁过期时间（需大于任务执行耗时，防止死锁）
    lock_expire = 10
    
    # 尝试获取锁：nx=True表示仅当key不存在时设置，ex=lock_expire设置自动过期
    if redis_client.set(lock_key, "locked", ex=lock_expire, nx=True):
        try:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 任务执行中 - Worker ID: {id(app)}")
            # 替换为你的实际任务逻辑
            time.sleep(2)
        finally:
            # 任务完成后释放锁
            redis_client.delete(lock_key)
    else:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 任务已被其他Worker执行 - Worker ID: {id(app)}")

@app.on_event("startup")
async def startup_event():
    # 配置APScheduler：Redis作为共享任务存储，确保多Worker任务列表一致
    jobstores = {
        'default': RedisJobStore(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
    }
    executors = {
        'default': ThreadPoolExecutor(20)
    }
    job_defaults = {
        'coalesce': False,
        'max_instances': 1
    }
    
    # 初始化异步调度器
    scheduler = AsyncIOScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)
    
    # 添加定时任务：每5秒执行一次，unique_task_id确保任务唯一
    scheduler.add_job(
        scheduled_task,
        trigger=IntervalTrigger(seconds=5),
        id="unique_task_id",
        replace_existing=True
    )
    
    scheduler.start()
    print("APScheduler 启动成功")

@app.get("/")
async def root():
    return {"message": "FastAPI + APScheduler 多Worker 部署示例"}

关键说明 #

• Redis JobStore：所有Worker共享同一个Redis任务存储，避免多实例启动时重复创建任务，保证任务配置全局一致。
• 分布式锁：通过Redis的set nx ex命令实现分布式锁，强制同一时间窗口内只有一个Worker执行任务，彻底避免重复触发。
• 任务参数配置：replace_existing=True确保多Worker启动时不会重复注册相同ID的任务；max_instances=1限制任务同时运行的实例数量。

部署测试 #

1. 启动多个Worker实例：

# 第一个Worker进程
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2
# 第二个Worker进程（模拟多节点部署）
uvicorn main:app --host 0.0.0.0 --port 8001 --workers 2

2. 查看控制台输出，会看到每5秒只有一个Worker执行任务，其余Worker会跳过本次执行。

内容的提问来源于stack exchange，提问作者Jhonathan Mizrahi