Django中select_for_update配合transaction.atomic()失效，多线程重复生成batch_id问题排查

你当前的代码之所以会出现重复batch_id，核心问题是select_for_update对聚合查询（aggregate(Max)）不起作用。select_for_update是行级锁，它只会锁定查询返回的具体行，但聚合操作是计算所有行的最大值，并不会锁定任何实际的数据行。多个并发线程进入事务后，都会同时拿到同一个最大batch_id，各自加1后保存，自然就产生了重复的ID。

方法一：用单独的序列记录表 #

创建一个专门存储当前batch_id最大值的模型，每次获取新ID时锁定唯一的一行，原子性更新并返回新值：

1. 定义序列模型：

from django.db import models

class BatchSequence(models.Model):
    current_id = models.IntegerField(default=0)

    @classmethod
    def get_next_batch_id(cls):
        with transaction.atomic():
            # 锁定唯一的序列行，不存在则自动创建
            seq, created = cls.objects.select_for_update().get_or_create(pk=1)
            seq.current_id += 1
            seq.save()
            return seq.current_id

2. 业务代码中调用：

json_dict['batch_id'] = BatchSequence.get_next_batch_id()
status_row = JobStatus(**json_dict)
status_row.save()

这种方式通过行级锁锁定序列表的唯一行，确保同一时刻只有一个线程能更新ID，彻底避免并发冲突。

方法二：利用数据库原生序列（适用于PostgreSQL等支持序列的数据库） #

如果你的数据库支持原生序列（比如PostgreSQL），可以直接依赖数据库的原子性序列生成能力：

1. 先在数据库中创建序列（可通过Django迁移脚本执行）：

CREATE SEQUENCE batch_id_seq START WITH 1 INCREMENT BY 1;

2. 在Django中调用获取序列值：

from django.db.models import Func, Value

class NextVal(Func):
    function = 'nextval'
    template = "%(function)s('%(sequence)s')"

# 获取下一个唯一的batch_id
next_batch_id = JobStatus.objects.aggregate(
    batch_id=NextVal(Value('batch_id_seq'))
)['batch_id']

json_dict['batch_id'] = next_batch_id
status_row = JobStatus(**json_dict)
status_row.save()

数据库原生序列的nextval操作是原子性的，每次调用都会返回唯一递增的ID，无需额外加锁，性能更高效。

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