任务调度引擎：Celery架构与部署

说到自动化运维，任务调度是绕不开的核心环节。你想想看，半夜三点爬起来执行脚本？那还不如让系统自己干活。Celery 就是干这个的——一个分布式任务队列，说白了就是「把任务扔进去，让工人排队干活」。

我个人习惯把 Celery 比作一个快递分拣中心。你（生产者）把包裹（任务）扔进传送带（消息队列），分拣员（Worker）按地址（路由）分发，快递车（结果后端）把签收单（结果）送回。嗯，这个比喻虽然糙，但道理不糙。

Celery 的核心架构

先看一张图，这是我画的核心架构图。别嫌丑，逻辑是清楚的。

这张图里，生产者就是你的 Django 或 Flask 应用。消息队列我用 Redis 比较多，轻量、快。Worker 是真正干活的进程，可以开多个。结果后端存返回值，方便你查任务跑完了没。

部署实战：从零跑通 Celery

我记得第一次部署 Celery 时踩了个坑——忘记启动 Worker，结果任务全堆在队列里，我还纳闷怎么没反应。后来养成了习惯：先看 Worker 日志，再谈其他。

下面是最简部署步骤。假设你用 Redis 做消息队列。

# 1. 安装依赖
pip install celery redis

# 2. 创建 tasks.py
from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0',
             backend='redis://localhost:6379/0')

@app.task
def add(x, y):
    return x + y

# 3. 启动 Worker（终端执行）
celery -A tasks worker --loglevel=info

# 4. 调用任务（Python 交互环境）
from tasks import add
result = add.delay(4, 5)
print(result.get())  # 输出 9

嗯，这里要注意：delay() 是异步调用，get() 会阻塞等待结果。生产环境别滥用 get()，容易把性能拖垮。

定时任务与周期任务配置

运维里最常用的就是定时任务。比如每天凌晨备份数据库、每小时清理日志。Celery 的 Beat 组件就是干这个的。

Beat 像一个闹钟，到点就自动把任务塞进队列。Worker 只管闷头干活，不用关心时间。

配置周期任务

有两种方式：硬编码在代码里，或者动态从数据库读取。我建议小项目用硬编码，大项目用数据库动态配置。

# 方式一：硬编码（适合小项目）
from celery import Celery
from celery.schedules import crontab

app = Celery('tasks', broker='redis://localhost:6379/0')

app.conf.beat_schedule = {
    'backup-db-every-day': {
        'task': 'tasks.backup_database',
        'schedule': crontab(hour=3, minute=0),  # 每天凌晨3点
        'args': (),
    },
    'clean-logs-every-hour': {
        'task': 'tasks.clean_logs',
        'schedule': 3600.0,  # 每3600秒执行一次
        'args': (),
    },
}

@app.task
def backup_database():
    print("备份数据库...")

@app.task
def clean_logs():
    print("清理日志...")

启动 Beat 组件：

celery -A tasks beat --loglevel=info

注意，Beat 和 Worker 是两个独立进程。我习惯用 Supervisor 同时管理它们，省心。

crontab 表达式详解

Celery 的 crontab 和 Linux 的 crontab 很像，但更灵活。看这个表格：

表达式	含义	示例
crontab()	每分钟执行	每分钟
crontab(minute=0, hour=0)	每天午夜	00:00
crontab(minute=0, hour='*/2')	每2小时	0点、2点、4点...
crontab(minute=30, hour=6, day_of_week=1)	每周一6:30	周一 06:30
crontab(minute='*/15')	每15分钟	0、15、30、45分

我个人习惯用 crontab 而不是 timedelta，因为 crontab 更精确。比如「每天凌晨3点」用 timedelta 很难表达，但 crontab 一行搞定。

任务链与错误重试机制

实际运维中，任务往往不是孤立的。比如：先下载文件，再解压，最后入库。这就需要用任务链把它们串起来。

任务链（Chain）

Celery 提供了 chain 方法，让任务按顺序执行。前一个任务的返回值，自动传给下一个任务。

from celery import chain

@app.task
def download_file(url):
    print(f"下载 {url}")
    return "/tmp/file.zip"

@app.task
def unzip_file(filepath):
    print(f"解压 {filepath}")
    return "/tmp/extracted"

@app.task
def import_data(dir_path):
    print(f"导入 {dir_path}")
    return "success"

# 任务链：下载 -> 解压 -> 导入
result = chain(
    download_file.s("http://example.com/data.zip"),
    unzip_file.s(),
    import_data.s()
)()

这里 .s() 是签名方法，相当于给任务预填参数。链式调用时，前一个任务的返回值会自动填充到后一个任务的第一个参数位置。

错误重试机制

任务总有失败的时候。网络超时、资源不足、数据异常……我见过最离谱的是磁盘满了，任务写日志都写不进去。

Celery 内置了重试机制，用 retry 参数控制。

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_alert(self, message):
    try:
        # 模拟发送告警
        print(f"发送告警: {message}")
        # 假设这里抛异常
        raise ConnectionError("网络不通")
    except Exception as exc:
        # 重试，指数退避
        countdown = 2 ** self.request.retries * 10
        raise self.retry(exc=exc, countdown=countdown)

关键参数说明：

参数	含义	建议值
max_retries	最大重试次数	3-5 次
default_retry_delay	默认重试间隔（秒）	60-300
countdown	自定义重试间隔	指数退避：2^retries * 10
autoretry_for	自动重试的异常类型	(ConnectionError, TimeoutError)

避坑指南：重试的陷阱

说几个我踩过的坑：

• 幂等性：重试的任务必须是幂等的。比如「扣钱」任务重试两次，用户就被扣了两次钱。我一般用唯一ID去重。
• 死信队列：重试超过最大次数后，任务会进入死信队列。记得监控死信队列，否则失败任务就石沉大海了。
• 任务超时：设置 soft_time_limit 和 time_limit，防止任务卡死。我习惯设 soft 为 300秒，hard 为 600秒。

@app.task(
    bind=True,
    max_retries=3,
    autoretry_for=(ConnectionError, TimeoutError),
    soft_time_limit=300,
    time_limit=600
)
def robust_task(self):
    # 这个任务会自动重试网络类异常
    pass

嗯，到这里 Celery 的核心用法基本讲完了。任务调度这东西，上手容易，但想跑得稳，得把重试、监控、幂等这些细节都考虑进去。我每次搭建新系统，都会先画架构图，再写代码——图清楚了，代码就不会乱。