Celery Chord串联Chain时任务异常引发数据库错误问题排查

架构老林

2026-7-28

问题描述

我使用AMQP作为任务队列，django_celery_results作为结果后端。当执行Chord任务串联Chain任务时，如果Chord的Group中有任务抛出异常，会触发数据库完整性错误；但仅使用Chord、或Group中无异常时执行完全正常。这是用法错误还是组件Bug？

任务代码

@shared_task(bind=True)
def debug_task(self):
    print(f'Request: debug_task')

@shared_task(bind=True)
def debug_A(self, i):
    print(f'Doing A{i}')

@shared_task(bind=True)
def debug_broke(self):
    print(f'OH NO')
    raise Exception("Head Aspload")

@shared_task(bind=True)
def debug_finisher(self):
    print(f'OK, We did it all')

def launch():
    from celery import group
    g = [debug_A.si(i) for i in range(5)]
    g.append(debug_broke.si())
    r = group(g) | debug_finisher.si() | debug_A.si(-1)
    r.apply_async()

核心错误日志

django.db.utils.IntegrityError: null value in column "task_id" violates not-null constraint
...

问题分析与解决方案

这是django_celery_results的已知Bug，并非用法错误。

原因

当Chord的Group内有任务失败时，Celery会自动生成chord_error任务处理异常，但django_celery_results在记录该错误任务的结果时，未正确填充task_id字段，触发了数据库的非空约束校验错误。

解决方案

升级到最新版django_celery_results：该问题在后续版本中已被修复，更新到最新稳定版即可彻底解决。

临时异常捕获方案：在Group内的任务中添加异常处理，避免抛出未捕获的异常。例如修改debug_broke任务：

@shared_task(bind=True)
def debug_broke(self):
    print(f'OH NO')
    try:
        raise Exception("Head Aspload")
    except Exception as e:
        # 返回错误信息而非直接抛出异常
        return str(e)