Celery的chord回调任务persist_task_es无报错却不执行如何解决？

排查步骤 #

• 确认结果后端配置正确
  Celery的chord回调依赖结果后端统计头部任务的完成状态，必须正确配置CELERY_RESULT_BACKEND参数，可选用Redis、Database、RPC等符合要求的后端，未配置或配置错误会直接导致回调不触发。
• 修正冗余的嵌套写法
  你当前代码中不需要在chord外层套空chain，chord本身支持直接调用delay()触发，修改后的代码如下：

# 原错误写法
# chained_tasks = chain(chord([taggers.s(text), entitied.s(text)], persist_task_es.s(user_task)))
# chained_tasks.delay()
# 修正后写法
chord(
    [taggers.s(text), entitied.s(text)], 
    persist_task_es.s(user_task)
).delay()

部分Celery 4.x版本存在chord嵌套chain的兼容性bug，移除外层chain可解决该类问题。

• 检查头部任务的结果存储配置
  如果全局配置了CELERY_IGNORE_RESULT = True，或者taggers、entitied两个任务单独设置了ignore_result=True，会导致任务执行结果无法写入结果后端，chord无法收到完成信号。可给两个头部任务单独关闭结果忽略：

@app.task(name='taggers', ignore_result=False)
def taggers(text: str) -> dict:
    # 原有逻辑

• 验证返回值可序列化
  确认get_tags、get_entities的返回值支持你配置的序列化协议（默认一般为json），如果存在无法序列化的对象（比如自定义类实例、二进制内容），任务执行后结果无法存储，会阻塞回调触发。可在两个任务结尾加日志打印返回值，确认执行无异常且返回值格式符合要求。
• 检查回调任务的消费配置
  1. 确认persist_task_es所属队列有对应的worker进程在消费，可通过Celery Flower工具查看队列积压情况
  2. 本地手动调用persist_task_es传入模拟参数，确认函数本身执行无报错，避免因为函数内部异常导致看起来未执行（注意检查日志级别配置，避免异常日志被过滤）

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