如何在Python Celery中创建不传递前置结果的任务链？

需求是构建一个异步任务链，让任务按顺序依次执行，但不传递前一个任务的结果给下一个，仅使用每个任务自身的参数。

现有代码如下：

构建任务链的代码：

tasks = []
for document_id in document_ids:
    tasks.append(
        utils.task(
            'tasks.update_document',
            document_id=document_id,
        ),
    )

result = chain(*tasks).apply_async(ignore_result=True)

任务定义代码：

@app.task
def update_document(document_id, *args,**kwargs):
    # 注：此处my_variable疑似笔误，应为document_id？
    if not isinstance(my_variable, int):
        raise TypeError
    return f'Done: {document_id}'

问题：设置ignore_result=True后，后续任务仍然接收到前一个任务的返回结果，导致执行失败。

问题原因 #

ignore_result=True的作用是让Celery不存储任务的执行结果，不会改变chain传递前序结果的默认行为。Celery的chain组件本身就是设计成将前一个任务的输出作为下一个任务的输入，因此即使不存储结果，参数传递逻辑依然生效。

解决方案 #

要避免参数传递，需要将每个任务定义为不可变签名（Immutable Signature），这样任务会忽略任何传入的额外参数，只使用自身预先定义的参数。

方法1：使用si快捷方式（推荐） #

si是signature(immutable=True)的简写，直接创建不可变任务签名：

from celery import si

tasks = []
for document_id in document_ids:
    # 用si创建不可变任务签名，指定自身参数
    tasks.append(
        si('tasks.update_document', document_id=document_id)
    )

# 执行任务链，无需设置ignore_result（若不需要存储结果仍可添加）
result = chain(*tasks).apply_async(ignore_result=True)

方法2：显式创建不可变Signature #

如果需要更精细的控制，可直接使用signature并设置immutable=True：

from celery import signature

tasks = []
for document_id in document_ids:
    tasks.append(
        signature(
            'tasks.update_document',
            kwargs={'document_id': document_id},
            immutable=True
        )
    )

result = chain(*tasks).apply_async(ignore_result=True)

效果说明 #

使用不可变签名后，任务链会按顺序执行每个update_document任务，每个任务只会使用自己的document_id参数，完全忽略前序任务的返回结果，解决参数传递导致的执行失败问题。

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