Django Celery操作Postgres频繁报out of shared memory错误如何解决

问题根因 #

你的代码没有锁未释放的问题，报错的核心原因是单事务内持有了过多行锁，超过了Postgres默认的max_locks_per_transaction阈值（默认值为64）：

• 你将整个遍历my_iterable的循环包裹在了同一个atomic事务块中，每次get_or_create、save操作都会对目标行加锁
• 所有行锁要等到整个大事务完全提交后才会释放，当my_iterable的元素量较大时，事务内累计持有的锁数量会快速超过阈值，触发共享内存不足的报错

解决方案 #

• 方案1：拆分大事务，批量提交 #

如果业务允许分批提交、不需要全量操作的原子性，可以把大事务拆分为小批量事务，控制每个事务内的锁数量：

import time
# 可根据实际场景调整批量大小，确保单事务锁数量远低于max_locks_per_transaction
BATCH_SIZE = 50
count_new_records = 0

for index, item in enumerate(my_iterable):
    # 每个元素单独开事务，或者攒满一批开一次事务，根据业务要求选择
    with transaction.atomic():
        template_obj, created = MyTable.objects.get_or_create(
            col1=col1, # 注意：你原代码中循环内的col1是外层固定值，可能是笔误，可根据实际逻辑调整
            defaults={
                'col2': col2,
                'col3': col3
            }
        )
        if not created:
            template_obj.col3 += col3
            template_obj.save()
        else:
            count_new_records += 1
    # 每处理完一批可短暂休眠，降低数据库压力
    if (index + 1) % BATCH_SIZE == 0:
        time.sleep(0.05)

• 方案2：用批量操作替代逐行操作（推荐） #

逐行读写数据库不仅锁多，性能也非常差，改为先批量查、再批量更新、最后批量插入的逻辑，整个流程只需要3次数据库交互，锁数量也会大幅降低：

from django.db import transaction

count_new_records = 0
# 先收集所有需要匹配的col1值（注意根据你的实际业务逻辑调整，原代码中col1为固定值的话可省略这步）
target_col1_list = [item对应col1值 for item in my_iterable]
# 批量查已存在的记录
exist_objs = MyTable.objects.filter(col1__in=target_col1_list)
exist_col1_map = {obj.col1: obj for obj in exist_objs}

to_update = []
to_create = []

for item in my_iterable:
    current_col1 = item对应col1值
    current_col2 = item对应col2值
    current_col3 = item对应col3值
    if current_col1 in exist_col1_map:
        obj = exist_col1_map[current_col1]
        obj.col3 += current_col3
        to_update.append(obj)
    else:
        to_create.append(MyTable(
            col1=current_col1,
            col2=current_col2,
            col3=current_col3
        ))

# 统一在一个事务内批量执行
with transaction.atomic():
    if to_update:
        MyTable.objects.bulk_update(to_update, fields=['col3'])
    if to_create:
        MyTable.objects.bulk_create(to_create)
        count_new_records = len(to_create)

• 兜底方案：适度调整数据库参数 #

如果业务确实需要大事务无法拆分，再考虑修改Postgres配置文件中的max_locks_per_transaction参数，调高阈值，修改后需要重启数据库生效。注意该参数不要调得过高，会占用更多共享内存。

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