FastAPI使用Gunicorn启动多worker时如何仅在单个worker运行定时任务

Gunicorn启动的多个worker是完全独立的进程，没有内置的进程间协调能力，所以必须通过外部锁、角色标记或者独立进程的方式实现定时任务仅单实例运行，以下是三种常用实现方案：

方案1：基于分布式锁实现（改造成本最低） #

• 选择通用中间件作为锁存储，推荐用Redis，数据库、Memcached也可实现
• 所有worker的定时任务触发时，先尝试获取全局唯一锁，只有成功拿到锁的worker才执行任务逻辑，执行完成后主动释放锁
• 必须给锁设置过期时间，避免worker异常崩溃导致锁永久无法释放
  Python+Redis实现示例：

import redis
from apscheduler.schedulers.background import BackgroundScheduler

redis_client = redis.Redis(host='你的Redis地址', port=6379, db=0)
# 全局唯一锁的key
LOCK_KEY = "scheduler_unique_lock"
# 锁过期时间要大于任务的最大预期执行时长，单位秒
LOCK_EXPIRE = 60

def your_scheduled_task():
    # nx参数表示只有key不存在时才设置成功，ex为过期时间
    get_lock = redis_client.set(LOCK_KEY, "running", ex=LOCK_EXPIRE, nx=True)
    if not get_lock:
        # 没拿到锁直接退出
        return
    try:
        # 此处写你的定时任务逻辑
        print("仅单worker执行的任务逻辑")
    finally:
        # 任务执行完成主动释放锁
        redis_client.delete(LOCK_KEY)

# 所有worker都可以正常初始化调度器，不会重复执行
scheduler = BackgroundScheduler()
scheduler.add_job(your_scheduled_task, 'cron', hour=2)
scheduler.start()

方案2：给worker加角色标记 #

• 不一次性启动4个worker，分开启动普通业务worker和定时任务专属worker，通过环境变量给专属worker打角色标记
• 代码中判断当前进程的环境变量，只有带定时任务角色标记的worker才会初始化启动调度器
  启动命令示例：

# 启动3个处理业务请求的普通worker
gunicorn main:app --workers 3 --bind 0.0.0.0:8000
# 单独启动1个带定时任务角色标记的worker
GUNICORN_ROLE=scheduler gunicorn main:app --workers 1 --bind 0.0.0.0:8000

代码判断逻辑：

import os
from apscheduler.schedulers.background import BackgroundScheduler

# 只有标记为定时任务角色的worker才会启动调度器
if os.getenv("GUNICORN_ROLE") == "scheduler":
    scheduler = BackgroundScheduler()
    scheduler.add_job(your_task, 'cron', minute=30)
    scheduler.start()

方案3：定时任务完全脱离服务进程部署 #

把定时任务逻辑拆成独立的代码片段，用系统crontab、独立的APScheduler进程、Celery Beat等组件单独部署运行，完全不和FastAPI的服务worker耦合，从根源上避免多实例重复执行的问题，适合定时任务逻辑较重的场景。

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