如何使用Celery链式任务统一入口并在beat_schedule中配置定时调度

现有代码问题梳理 #

你的示例代码存在3个核心问题，导致无法实现预期的链式调度逻辑：

• 使用si()不可变签名：该签名不会将前序任务的返回值作为参数传递给下一个任务，无法实现任务间的依赖传值
• 链式入口是普通函数：Celery Beat 只能调度被@app.task装饰器注册的任务，普通函数无法直接加入定时配置
• 参数不匹配：mul_demo定义要求2个入参，但你调用时只传了1个参数，运行时会触发参数错误

具体实现步骤 #

1. 修正任务签名与链式逻辑 #

Celery 的chain本身就具备串行等待能力，前序任务执行完成（成功返回）才会自动触发下一个任务，无需额外实现类似asyncio.join的逻辑。如果需要传递前序任务返回值，使用普通的.s()签名即可，不需要.si()。

2. 封装链式入口为Celery任务 #

将组装并执行链式任务的逻辑封装为标准Celery任务，供Beat调度。

3. 修改Beat配置加入定时触发规则 #

将封装好的链式任务加入beat_schedule配置，实现定时运行。

完整可运行示例代码 #

from celery import Celery, chain
from celery.schedules import crontab, timedelta
import time

# 请根据实际情况修改broker和backend配置
app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

# 定时调度配置
app.conf.beat_schedule = {
    # 每2分钟运行一次链式任务
    'run-chain-every-2min': {
        'task': 'tasks.run_task_chain',
        'schedule': timedelta(minutes=2),
    },
    # 每天8:50运行一次链式任务
    'run-chain-at-850': {
        'task': 'tasks.run_task_chain',
        'schedule': crontab(hour=8, minute=50),
    }
}

@app.task
def add_demo(x, y):
    time.sleep(3)
    return x + y

@app.task
def mul_demo(x, y):
    time.sleep(3)
    return x * y

@app.task(ignore_result=True)
def insert_db_demo(result):
    print('insert db , result {}'.format(result))

# 链式任务入口，注册为Celery任务
@app.task(bind=True, ignore_result=True)
def run_task_chain(self):
    task_chain = chain(
        add_demo.s(10, 7), # 执行后返回17，自动传给下一个任务
        mul_demo.s(10), # 接收前序返回的17作为第一个参数，等价于mul_demo(17, 10)，返回170
        insert_db_demo.s() # 接收前序返回的170作为参数
    )
    # 触发链式任务执行
    task_chain.apply_async()

关键注意事项 #

• 必须配置Celeryresult_backend（比如Redis、数据库）：链式任务依赖前序任务的返回结果，没有结果后端无法传递参数
• 中间任务不要加ignore_result=True：如果是链中间的任务开启了忽略结果，后序任务会拿不到前序返回值
• 异常处理：可以给链式任务追加link_error回调，处理任意节点执行失败的场景，例如：

  task_chain.apply_async(link_error=handle_chain_error.s())
  

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