Django 4.0批量更新模型并避免全表锁的可行方法

由于直接执行MyModel.objects.all().update(touched=now())会触发全表锁，针对大数量场景，推荐以下两种分批更新的方案：

一、按主键分段手动分批更新（最稳妥高效） #

利用主键的连续性，将全表数据拆分成多个小批次，每个批次在独立事务中执行更新，既能缩小锁的范围，也能避免长时间占用表锁。

实现示例（Django管理命令） #

from django.core.management.base import BaseCommand
from django.utils import timezone
from myapp.models import MyModel
from django.db import transaction, models

class Command(BaseCommand):
    help = '分批更新MyModel的touched字段为当前时间，避免全表锁'

    def add_arguments(self, parser):
        parser.add_argument('--batch-size', type=int, default=1000, help='每批次处理的记录数量，默认1000')

    def handle(self, *args, **options):
        batch_size = options['batch_size']
        now = timezone.now()

        # 获取主键的最小、最大值，确定更新范围
        id_range = MyModel.objects.aggregate(
            min_id=models.Min('id'),
            max_id=models.Max('id')
        )
        min_id, max_id = id_range['min_id'], id_range['max_id']

        if not min_id or not max_id:
            self.stdout.write(self.style.SUCCESS('无待更新记录'))
            return

        current_start = min_id
        while current_start <= max_id:
            current_end = current_start + batch_size - 1
            # 确保最后一批不超出最大ID
            current_end = min(current_end, max_id)

            # 每个批次用独立事务执行
            with transaction.atomic():
                updated_count = MyModel.objects.filter(
                    id__gte=current_start,
                    id__lte=current_end
                ).update(touched=now)
            
            self.stdout.write(
                self.style.SUCCESS(f'完成ID范围 [{current_start}, {current_end}] 的更新，共处理{updated_count}条记录')
            )
            current_start = current_end + 1

        self.stdout.write(self.style.SUCCESS('所有记录更新完成'))

二、使用iterator()分批加载后更新 #

如果需要触发模型的save()信号或有额外业务逻辑，可通过iterator(chunk_size=...)分批加载对象，再用bulk_update批量保存：

from django.utils import timezone
from myapp.models import MyModel

now = timezone.now()
batch = []
chunk_size = 1000

for obj in MyModel.objects.all().iterator(chunk_size=chunk_size):
    obj.touched = now
    batch.append(obj)
    # 达到批次大小就执行批量更新
    if len(batch) >= chunk_size:
        MyModel.objects.bulk_update(batch, ['touched'])
        batch = []
# 处理剩余的最后一批
if batch:
    MyModel.objects.bulk_update(batch, ['touched'])

注意：bulk_update不会触发pre_save/post_save信号，适合纯字段值更新的场景，性能远高于单条save()。

关键注意点 #

• 批次大小建议根据数据库性能调整，通常1000-5000条是比较合理的范围：太小会增加数据库请求次数，太大仍可能导致锁表时间过长
• 优先选择主键作为分段条件（主键自带索引，查询效率高，锁范围精准）；如果无自增主键，可改用其他带索引的字段（如created_at）按区间拆分
• 执行过程中加入日志输出，方便监控进度和排查异常

内容的提问来源于stack exchange，提问作者Mr T.