Celery chain 与 group 组合后任务重复执行、chord_unlock 重试如何排查？

2026-11-04数据小陈

结论

日志已经证实任务发生了重复执行，但还不能确定重复投递或重复发布的根因。.si() 只控制参数传递，不能修复重复执行；在当前 PostgreSQL result_backend 下，celery.chord_unlock 的短暂重试可能是正常的等待行为。

如果“一个任务完成”还包括它的 task_two 回调完成，建议把 task_one → task_two 写成一条 chain，再用 chord 等待整批任务完成。这样能明确表达依赖关系，但不能保证消除现有环境中的重复执行。

先区分执行顺序与重复执行

后一次测试的日志显示，前两个批次大致每隔 10 秒推进一次，与 task_one 中的 time.sleep(10) 设置相符。需要追查的是后续的重复记录：

a749cc89-b0ec-4bed-adad-95247efd8f68 对应的第 5 个任务被接收两次，并在两个进程中分别完成。
第 6 个任务也以相同的 task_id 完成两次，后续第 7、8 个任务继续重复。
最终，同一个 barrier task_id 被接收两次，随后发生结果表唯一约束冲突。

这说明任务确实重复执行，而不只是日志交错。单独一条 received 记录只能说明任务被接收，不能证明它已经开始运行；同一 ID 在不同进程中分别出现一次 succeeded，则是重复执行的证据。

`.si()`、`link` 和批次等待分别做什么？

group 表达组内并行；需要等待整组完成后继续时，可以使用 chord。group 后连接其他任务的常见 Canvas 写法也会形成 chord 依赖，因此不能直接判定原来的组合方式有误。

.s() 可以接收上游结果，.si() 会忽略追加的上游参数，但仍保留执行依赖。给第二批的 task_one 使用 .si(i)，可以避免第一批的结果作为额外参数传入 task_one(i)。Celery Canvas 文档

原代码使用的是：

task_one.si(i).set(link=task_two.s(i))

这里 task_two 是成功回调。如果批次中只有 task_one，回调完成就不属于该批次的完成条件。业务要求等待回调时，可以把它加入批次中的任务链：

chain(task_one.si(i), task_two.s(i))

这样，task_two 的结果才是该分支最终用于批次汇合的结果。

用明确的批次边界组织任务

下面的示例沿用已经初始化的 app 和 QUEUE。每个批次包含两个分支，每个分支依次执行 task_one 和 task_two。整批分支成功完成后执行 barrier，然后进入下一批。

import time
from uuid import uuid4

from celery import chain, chord, group


@app.task(ignore_result=False)
def task_one(i):
    time.sleep(10)
    return f"The {i}th task was executed successfully."


@app.task(ignore_result=False)
def task_two(result, task_id):
    return f"Execution completed:{result}, task_id:{task_id}"


@app.task(ignore_result=False)
def barrier(x):
    return f"Midpoint：{x}"


run_id = uuid4().hex
batches = [[1, 2], [3, 4], [5, 6], [7, 8]]

# 可按任务分别修改这两类队列；worker 需要消费对应队列。
queues = {
    i: {"one": QUEUE, "two": QUEUE}
    for batch in batches
    for i in batch
}


def make_branch(i):
    one_id = f"{run_id}:task_one:{i}"
    two_id = f"{run_id}:task_two:{i}"

    return chain(
        task_one.si(i).set(
            queue=queues[i]["one"],
            task_id=one_id,
        ),
        task_two.s(one_id).set(
            queue=queues[i]["two"],
            task_id=two_id,
        ),
    )


def make_batch(batch_no, numbers):
    return chord(
        group([make_branch(i) for i in numbers]),
        barrier.s().set(
            queue=QUEUE,
            task_id=f"{run_id}:barrier:{batch_no}",
        ),
    )


workflow = chain(*[
    make_batch(batch_no, numbers)
    for batch_no, numbers in enumerate(batches, start=1)
])

result = workflow.apply_async()

示例中的参数和结果传递方式如下：

分支起点使用 task_one.si(i)，忽略上一批传来的结果。
task_two.s(one_id) 保留结果注入，实际调用形式为 task_two(result, one_id)。
barrier 接收当前批次各分支的结果，在这个示例中是 task_two 的结果列表。
最后的 barrier 只接收最后一批的结果；如果需要汇总所有批次，应另行设计结果汇总。

run_id 用于区分不同的工作流实例。每次主动发起新的工作流时，都应重新构造签名和 ID。run_id 不是去重机制，也不保证任务只执行一次。

另外，task_two.s(5) 中的 5 只是函数参数，并没有设置 Celery 消息的 task_id。消息 ID 是通过 .set(task_id=...) 设置的。

`chord_unlock` 每秒重试是否说明执行失败？

当前配置的 result_backend 是 PostgreSQL。broker 使用 Redis，并不意味着结果汇合也使用 Redis 后端。

对于采用轮询方式等待结果的后端，celery.chord_unlock 会检查 header 是否完成；如果尚未完成，就通过 retry 再次检查。检查结果状态时发生异常，也可能触发重试。因此，单独一行 Retry in 1.0s 不能证明业务任务失败。Celery chord_unlock 实现

这份日志中，相关任务约在 11:09

被接收，11:09

完成。期间 unlock 重试，随后也成功结束，与等待约 10 秒的任务完成相符。

如果业务任务已经结束，unlock 仍长期重试，应检查对应任务的结果状态、数据库访问异常、结果是否被忽略，以及生产者和 worker 是否使用相同的 result_backend。参与 chord 的任务需要保存结果，示例通过 ignore_result=False 明确这一点。Celery chord 注意事项

如何定位重复执行的来源？

建议从最小复现开始，每次只增加一层结构：

记录 Celery、Kombu、Redis 客户端、SQLAlchemy、psycopg2 的实际版本，以及 worker 启动参数和生效配置。当前材料没有提供 Celery 版本，不能据此归因于某个版本缺陷。
使用单一提交入口，重新构造工作流并只提交一次；先测试两个批次，暂时移除 link。
为每次提交记录工作流标识，为任务记录参数、self.request.id、root_id、parent_id、retries 和 delivery_info，并增加实际开始执行时的日志。
对照生产者发布日志与 worker 接收日志，区分同一逻辑任务被多次发布，还是消息被重新投递。若依赖图在一次发布过程中重复展开，再针对实际版本检查 Canvas 行为。
恢复 task_one → task_two 分支，确认重复从哪一层开始出现。

原始代码使用固定日期和序号构造 ID，多次提交时可能复用，需要调整。不过，后一次代码已经没有手动设置 task_id，日志中仍出现相同 UUID 的重复执行，因此不能把所有问题都归因于固定 ID。

配置中的 visibility_timeout 是 3600 秒，示例中的重复接收则发生在很短时间内。仅凭这份日志，没有依据将该超时参数认定为本次重复的原因。

PostgreSQL 唯一约束冲突怎么处理？

日志显示，两个 barrier 执行实例使用同一个 task_id，并在相近时间写入结果表，随后触发 celery_taskmeta_task_id_key 冲突。这支持“重复执行引发并发结果写入竞争”的判断，但不能说明最初是谁重复发布或投递了任务。

应保留唯一约束，优先追查重复执行的来源。删除约束或清空结果表，都无法建立正确的任务依赖，也无法阻止重复的业务操作。

对于写数据库、调用外部接口等有副作用的任务，还应按业务操作标识实现幂等处理。Celery 的 task_id 可以用于追踪，但不应单独承担业务去重职责。