Celery异步任务中双计数器同步递增异常的原因排查求助

问题场景 #

单个接口接收多并发图片上传请求，上传成功后需递增两个计数器：Album.image_count_in_album和Event.number_of_images_across_albums。同步递增因竞态条件出错，改用Celery异步任务后，仅number_of_images_across_albums计数正确，image_count_in_album仍存在错误。使用Postgres数据库。

代码与模型 #

异步任务代码 #

@shared_task
def increment_album_and_event_image_count(album_id):
    try:
        with transaction.atomic():
            album = Album.objects.get(id=album_id)
            album.image_count_in_album = F("image_count_in_album") + 1
            album.save()
            album.event.number_of_images_across_albums = (
                F("number_of_images_across_albums") + 1
            )
            album.event.save()
        return True
    except Exception as e:
        logger.error(f"Error incrementing image count: {e}, album: {album}")
        return False

模型定义 #

class Album(TimeStampedModel):
    image_count_in_album = models.IntegerField(default=0)
    event = models.ForeignKey("Event", on_delete=models.CASCADE)

class Event(TimeStampedModel):
    number_of_images_across_albums = models.IntegerField(default=0)

根本原因分析 #

看似相同的F表达式更新，结果却不一致，核心问题在于Django实例缓存与事务行锁的交互：

1. 虽然F表达式本身是数据库层面的原子操作，但在多任务并发场景下，若未显式加锁，Album.objects.get(id=album_id)可能读取到事务未提交的旧值（Postgres默认READ COMMITTED隔离级别下，语句执行时读取已提交数据，但任务并发时仍可能出现时序问题）。
2. 你通过album.event获取关联对象时，用的是Django实例缓存的Event对象，而非重新查询数据库，但这本身不影响F表达式的更新——真正的差异可能来自：Event是多Album共享的对象，其更新的行锁竞争更明显，反而迫使任务排队执行；而Album的更新因任务调度或事务提交时机，出现了部分任务的更新未被正确持久化。
3. 另一个潜在可能：TimeStampedModel的save()逻辑干扰了image_count_in_album的F表达式更新，导致部分任务的递增操作未生效。

解决方法 #

方法1：显式加行锁确保原子性 #

在获取实例时添加行锁，强制任务排队执行，彻底避免竞态：

@shared_task
def increment_album_and_event_image_count(album_id):
    try:
        with transaction.atomic():
            # 对Album加排他行锁，阻塞其他事务的修改
            album = Album.objects.select_for_update().get(id=album_id)
            album.image_count_in_album = F("image_count_in_album") + 1
            album.save()
            # 对关联Event也加锁，确保整体更新原子性
            event = Event.objects.select_for_update().get(id=album.event.id)
            event.number_of_images_across_albums = F("number_of_images_across_albums") + 1
            event.save()
        return True
    except Exception as e:
        logger.error(f"Error incrementing image count: {e}, album_id: {album_id}")
        return False

方法2：直接用数据库更新跳过实例缓存 #

避免实例缓存的干扰，直接通过QuerySet执行原子更新，效率更高：

@shared_task
def increment_album_and_event_image_count(album_id):
    try:
        with transaction.atomic():
            # 直接更新Album，无需获取实例
            Album.objects.filter(id=album_id).update(image_count_in_album=F("image_count_in_album") + 1)
            # 通过关联查询更新Event
            event_id = Album.objects.values_list('event_id', flat=True).get(id=album_id)
            Event.objects.filter(id=event_id).update(
                number_of_images_across_albums=F("number_of_images_across_albums") + 1
            )
        return True
    except Exception as e:
        logger.error(f"Error incrementing image count: {e}, album_id: {album_id}")
        return False

方法3：移除冗余计数器（推荐） #

如果业务允许，直接通过关联查询实时统计数量，彻底避免计数不一致：

• 替代image_count_in_album：用album.image_set.count()（假设图片模型与Album为外键关联）
• 替代number_of_images_across_albums：用Image.objects.filter(album__event=event).count()

这种方式虽然每次查询需要计算，但数据绝对准确，适合对实时性要求不是极高的场景。

验证建议 #

1. 检查Celery任务日志，确认是否有任务执行失败（尤其是Album更新环节）
2. 直接查询数据库中的image_count_in_album值，不要依赖Django实例缓存或admin显示的数据
3. 用Locust等工具模拟高并发上传，对比实际上传数量与计数器值

内容的提问来源于stack exchange，提问作者hemant kumar chaudhary