Celery 4.10任务链技术问询：多任务间结果传递实现

针对你的需求，我们需要让三个任务按顺序执行，并且第三个任务能同时拿到第一个和第二个任务的结果。下面分两种场景给出实用的实现方案：

方案一：修改任务返回值（推荐，简洁优雅） #

这种方式通过让第二个任务返回包含前序结果的元组，直接把需要的结果传递给第三个任务，不需要依赖额外的Celery特性，是最直观的实现方式。

步骤说明： #

1. 调整register_user的返回值：让它同时返回validate_result和自身生成的registration_result，这样后续任务就能一次性拿到两个关键结果。
2. 调整update_user的参数处理：接收这个元组并解包，再结合给定的parameter_2执行业务逻辑。
3. 构建任务链：按顺序串联三个任务的签名即可，Celery会自动把前一个任务的结果作为下一个任务的第一个参数。

完整代码示例： #

from celery import Celery

# 根据你的实际环境配置Celery
app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

parameter_1 = "This is a parameter"
parameter_2 = "This is another parameter"

@app.task
def validate_user(user_credentials):
    # 模拟用户验证逻辑
    validate_result = {"valid": True, "user_id": 123}
    return validate_result

@app.task
def register_user(validate_result, param1):
    # 模拟用户注册逻辑
    registration_result = {"status": "success", "username": "test_user"}
    # 返回包含前序结果和当前结果的元组
    return (validate_result, registration_result)

@app.task
def update_user(results_tuple, param2):
    # 解包拿到两个前序任务的结果
    validate_result, registration_result = results_tuple
    # 模拟用户信息更新逻辑
    update_result = {
        "user_id": validate_result["user_id"],
        "username": registration_result["username"],
        "updated_field": param2
    }
    return update_result

# 构建并执行任务链
user_credentials = {"username": "test", "password": "123456"}
task_chain = chain(
    validate_user.s(user_credentials),
    register_user.s(parameter_1),
    update_user.s(parameter_2)
)

# 异步执行任务链
result = task_chain.apply_async()
# 获取最终执行结果
final_result = result.get()
print(final_result)

方案二：不修改任务返回值（依赖Result Backend） #

如果不想改动已有任务的返回值，可以通过Celery的任务上下文（self.request）和Result Backend来回溯获取前面任务的结果。这种方式需要任务绑定（bind=True），且要求你已经配置好有效的Result Backend（如Redis、MongoDB）。

完整代码示例： #

from celery import Celery
from celery.result import AsyncResult

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

parameter_1 = "This is a parameter"
parameter_2 = "This is another parameter"

@app.task
def validate_user(user_credentials):
    validate_result = {"valid": True, "user_id": 123}
    return validate_result

@app.task
def register_user(validate_result, param1):
    registration_result = {"status": "success", "username": "test_user"}
    return registration_result

@app.task(bind=True)
def update_user(self, registration_result, param2):
    # 通过任务上下文获取父任务（register_user）的ID
    parent_task_id = self.request.parent_id
    # 回溯获取validate_user任务的ID
    grandparent_task = AsyncResult(parent_task_id)
    validate_task_id = grandparent_task.request.parent_id
    # 从Result Backend中获取validate_user的结果
    validate_result = AsyncResult(validate_task_id).get()
    
    # 模拟用户信息更新逻辑
    update_result = {
        "user_id": validate_result["user_id"],
        "username": registration_result["username"],
        "updated_field": param2
    }
    return update_result

# 构建并执行任务链
user_credentials = {"username": "test", "password": "123456"}
task_chain = chain(
    validate_user.s(user_credentials),
    register_user.s(parameter_1),
    update_user.s(parameter_2)
)

result = task_chain.apply_async()
final_result = result.get()
print(final_result)

关键注意事项： #

• 使用.s()而非.si()：.si()是「不可变签名」，不会自动传递前一个任务的结果；我们需要前序结果作为后续任务的参数，所以必须用.s()。
• Result Backend配置：方案二依赖Result Backend存储历史任务结果，确保你的Celery配置中正确设置了backend参数。
• 异常处理：生产环境中建议为每个任务添加异常捕获逻辑，可结合link_error处理任务链中的失败场景，避免整个流程中断。

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