使用MySQL InnoDB的Django项目中，并发DML操作如何保证数据正确性？

你的核心问题是两个进程并发操作时，读-改-写的非原子性和循环更新的非原子性导致的数据异常，结合MySQL InnoDB的特性，直接从代码层面修复即可：

一、修复插入新记录的逻辑：确保原子性 #

原来的先查最大值再加1再插入是拆分的三步，并发下会出现多个进程拿到相同最大值、插入重复queue值的问题。可以用两种方式解决：

方式1：用Django事务+原子查询 #

通过transaction.atomic()包裹查询和插入操作，InnoDB会自动加锁，确保这一系列操作不会被其他进程打断：

from django.db import transaction

# 包裹在事务中，确保查询max和插入是原子操作
with transaction.atomic():
    # 聚合查询当前最大queue值，空表时max_queue为None
    max_queue_result = Task.objects.all().aggregate(max_queue=Max('queue'))
    queue_number = (max_queue_result['max_queue'] or 0) + 1
    # 插入新记录
    Task.objects.create(task_id=task_id, queue=queue_number)

方式2：用原生SQL直接原子插入 #

跳过Python层面的查询，直接让数据库计算最大值并插入，这是最高效的原子操作：

from django.db import connection

# 替换yourapp为你的Django应用名称，Task模型对应的表名默认是appname_task
with connection.cursor() as cursor:
    cursor.execute("""
        INSERT INTO yourapp_task (task_id, queue)
        VALUES (%s, COALESCE((SELECT MAX(queue) FROM yourapp_task), 0) + 1)
    """, [task_id])

二、修复批量减queue的逻辑：用原子批量更新 #

原来的循环逐个更新不仅效率低，还会导致在循环过程中插入的新非0queue记录被漏掉。改用Django的F()表达式直接在数据库层面完成批量更新，这是原子操作，不会被打断：

from django.db.models import F

# 直接更新所有queue≠0的记录，queue字段减1，数据库一次性完成操作
Task.objects.filter(queue__ne=0).update(queue=F('queue') - 1)

注意：你原来的代码里写了~Q(task_queue=0)，但模型字段是queue，这里要修正为正确的字段名

三、极端高并发场景的额外处理 #

如果并发量极大，担心插入时出现锁冲突或唯一键冲突（如果给queue加了unique=True约束），可以加异常捕获并重试：

from django.db import transaction, IntegrityError

while True:
    try:
        with transaction.atomic():
            max_queue_result = Task.objects.all().aggregate(max_queue=Max('queue'))
            queue_number = (max_queue_result['max_queue'] or 0) + 1
            Task.objects.create(task_id=task_id, queue=queue_number)
        break
    except IntegrityError:
        # 并发插入导致重复queue，重新计算最大值并重试
        continue

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