Django与Celery集成时出现DoesNotExists Error问题求助

咱们先来看你的代码和遇到的报错情况：

原代码 #

@app.task
def t_select(booking_id, seat_id, user_id):
    with transaction.atomic():
        booking = Booking.objects.get(id=booking_id)
        booking.seats.add(Seat.objects.get(pk=seat_id))
        return booking.save()

报错日志 #

[2018-04-01 02:25:55,893: INFO/MainProcess] Received task: functors.booker.t_select[cec7f2ca-bca0-4f3e-9865-102f3528ce3a]
[2018-04-01 02:25:55,894: ERROR/ForkPoolWorker-4] Task functors.booker.t_select[e3a69d6b-1fc9-44cb-96ce-bfe18a1c96a9...]

问题分析 #

从代码和报错来看，最可能的几个触发点：

1. 未处理的对象不存在异常：Booking.objects.get()或Seat.objects.get()如果找不到对应ID的记录，会直接抛出DoesNotExist异常，Celery任务没做捕获的话就会直接失败。
2. 多余的save()调用：Django中ManyToMany字段的add()方法会自动保存关联关系，手动调用booking.save()完全没必要，甚至可能引发冗余的数据库操作。
3. 缺少重试机制：如果是临时的数据库连接波动导致的问题，没有重试逻辑会直接导致任务失败。

修复后的代码 #

from django.db import transaction, ObjectDoesNotExist
from celery import app

@app.task(bind=True, retry_backoff=3)
def t_select(self, booking_id, seat_id, user_id):
    try:
        with transaction.atomic():
            # 先获取两个对象，后续统一捕获不存在的异常
            booking = Booking.objects.get(id=booking_id)
            seat = Seat.objects.get(pk=seat_id)
            
            # 检查座位是否已经在订单里，避免重复添加操作
            if seat not in booking.seats.all():
                booking.seats.add(seat)
                return f"座位{seat_id}已成功添加到订单{booking_id}"
            else:
                return f"座位{seat_id}已经在订单{booking_id}中了"
    
    # 捕获对象不存在的异常，自动重试最多2次，间隔自动递增
    except ObjectDoesNotExist as e:
        self.retry(exc=e, max_retries=2)
    
    # 兜底处理其他未知异常，返回错误信息方便排查
    except Exception as e:
        return f"任务执行失败：{str(e)}"

额外优化建议 #

• 高并发场景下，建议给查询加上select_for_update()，避免出现竞态条件（比如多个任务同时抢同一个座位）：

  booking = Booking.objects.select_for_update().get(id=booking_id)
  seat = Seat.objects.select_for_update().get(pk=seat_id)
  

• 确保Celery Worker的数据库配置和Django主应用完全一致，避免连接到错误的数据库实例。
• 可以配置Celery的CELERY_DB_REUSE_MAX参数，限制数据库连接的复用次数，避免长时间运行的Worker出现连接超时问题。

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