Celery Beat 定时调度配置

示例说明

定时调度的痛点不在触发而在编排：全部任务挤同一队列会互相阻塞，同一时刻齐发会打瘫下游。队列路由把账单、维护、报表分流到专属 worker；crontab 语义覆盖分钟步进、星期与月日组合；expires 选项让滞留的调度在过期后自动作废——worker 长时间宕机恢复后不会补跑一堆过时任务。

技术要点

• 队列路由隔离不同类别负载
• crontab 表达星期与月日组合
• expires 防止宕机后补跑过时任务

代码解读

四个调度条目分别演示：凌晨两点关账、每十五分钟缓存预热、周一六点半欧洲区报表（带参数）、每月一日合规导出（带过期窗口）。时区统一设 UTC 防多机漂移，任务函数按队列声明各自归属。

使用建议

关键任务加抖动或锁防与其他系统整点撞车。beat 是单点，高可用需双实例加锁或改用分布式调度器。调度与任务的队列声明要一致。

代码示例

from celery import Celery
from celery.schedules import crontab

app = Celery('ops', broker='redis://localhost:6379/1')
app.conf.update(
    timezone='UTC',
    beat_schedule={
        'nightly-invoice-close': {
            'task': 'tasks.close_invoices',
            'schedule': crontab(hour=2, minute=0),
            'options': {'queue': 'billing'},
        },
        'cache-warm-every-15min': {
            'task': 'tasks.warm_cache',
            'schedule': crontab(minute='*/15'),
            'options': {'queue': 'maintenance'},
        },
        'staggered-region-reports': {
            'task': 'tasks.region_report',
            'schedule': crontab(hour=6, minute=30, day_of_week='mon'),
            'args': ('eu',),
            'options': {'queue': 'reporting'},
        },
        'monthly-compliance-export': {
            'task': 'tasks.compliance_export',
            'schedule': crontab(day=1, hour=3, minute=0),
            'options': {'queue': 'reporting', 'expires': 7200},
        },
    },
)


@app.task(queue='billing')
def close_invoices() -> dict:
    return {'closed': 42}


@app.task(queue='maintenance')
def warm_cache() -> dict:
    return {'warmed': ['rates', 'catalog']}


@app.task(queue='reporting')
def region_report(region: str) -> dict:
    return {'region': region, 'rows': 128}


@app.task(queue='reporting')
def compliance_export() -> dict:
    return {'archive': 'compliance-2026-10.zip'}