Gunicorn多进程环境下如何让定时任务仅在单个Worker运行？

一、让定时任务仅在单个Gunicorn Worker中运行 #

从根源上避免多Worker重复执行任务，是最直接的解决方式。

方案1：通过Worker ID判断启动定时任务 #

Gunicorn会为每个Worker设置环境变量WORKER_ID（从0开始编号），我们可以只让编号为0的Worker启动APScheduler：

from flask import Flask
from flask_apscheduler import APScheduler
import os

app = Flask(__name__)
scheduler = APScheduler()

def your_scheduled_task():
    # 这里写你的任务逻辑：查找未处理条目并处理
    pass

# 仅在生产环境的第一个Worker中启动定时任务
if __name__ != '__main__':
    worker_id = os.environ.get('WORKER_ID')
    if worker_id == '0':
        scheduler.init_app(app)
        # 配置定时任务：每Y小时执行一次
        scheduler.add_job(
            id='unique_processing_task',
            func=your_scheduled_task,
            trigger='interval',
            hours=Y
        )
        scheduler.start()

注意事项：

• 如果使用gunicorn --preload参数，要确保数据库连接等资源在Worker进程中重新初始化，避免主进程的连接被多个Worker共享导致异常。
• 若Worker 0意外重启，重启后的Worker 0会重新接管定时任务，不会出现任务中断。

方案2：将定时任务独立于Gunicorn Worker运行 #

把定时任务完全抽离出来，不依赖Gunicorn的Worker进程，更稳定可靠：

• 写一个单独的Python脚本，直接初始化APScheduler并运行定时任务，和Flask应用分开启动。
• 或者用系统定时服务（如Linux的cron），每隔Y小时调用Flask应用的一个内部处理接口（记得给接口加身份验证，比如Token校验，避免非法调用）。

二、通过数据库原子操作实现任务幂等性 #

如果无法避免多Worker同时执行任务，通过数据库的原子操作可以保证同一条目不会被重复处理，从数据层面避免竞态。

方案1：使用SELECT ... FOR UPDATE SKIP LOCKED（推荐） #

如果你的数据库支持该语法（如PostgreSQL、MySQL 8.0+），可以在查询未处理条目时锁定数据，其他Worker会自动跳过已锁定的条目：

from your_app import db
from your_models import Entry

def your_scheduled_task():
    # 原子性获取未处理条目并锁定，其他Worker无法读取已锁定的条目
    with db.session.begin():
        unprocessed_entries = db.session.query(Entry)\
            .filter_by(processed=False)\
            .with_for_update(skip_locked=True)\
            .all()
        
        for entry in unprocessed_entries:
            # 执行你的处理逻辑
            entry.processed = True
    # 事务自动提交，释放锁

方案2：原子更新+判断执行 #

通过原子更新processed字段，只有更新成功（说明条目之前未被处理）的Worker才会执行后续逻辑：

from your_app import db
from your_models import Entry

def process_single_entry(entry_id):
    # 原子更新：仅当processed为False时才改为True，返回修改行数
    updated_count = db.session.query(Entry)\
        .filter(Entry.id == entry_id, Entry.processed == False)\
        .update({'processed': True}, synchronize_session=False)
    
    db.session.commit()
    
    if updated_count == 1:
        # 更新成功，执行该条目的处理逻辑
        pass

def your_scheduled_task():
    # 获取所有未处理条目的ID
    unprocessed_ids = [entry.id for entry in db.session.query(Entry.id).filter_by(processed=False).all()]
    
    for entry_id in unprocessed_ids:
        process_single_entry(entry_id)

总结 #

• 优先选择让单个Worker执行定时任务的方案，从根源上避免竞态；
• 若任务独立性强，推荐独立运行定时任务，降低与Web服务的耦合；
• 若需要兼容多Worker场景，使用数据库原子操作保证任务幂等性，即使多Worker同时执行也不会重复处理数据。

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