使用Django bulk_create时出现PostgreSQL内存泄漏问题

我写了一段定时拉取API数据并写入PostgreSQL的代码，每次循环都会在postgresql\operations.py文件的cursor.query.decode()处出现持续加剧的内存泄漏，即使手动调用gc.collect()和reset_queries()也无法解决，仅在执行bulk_create时触发该问题。

代码示例 #

import json
import time
import gc
from django.db import models

class DataObject(models.Model):
    raw_data = models.TextField()
    name = models.TextField(null=True)
    
    def post_process(self):
        data = json.loads(self.raw_data)       
        self.name = data['name']
        
def do_from_schedule():
    api = API()
    loads = api.grab_all_data()
    del api

    datas_to_create = []

    for load in loads:
        data = DataObject()
        data.raw_data = json.dumps(load)
        data.post_process()
        datas_to_create.append(data)

    DataObject.objects.bulk_create(datas_to_create, ignore_conflicts=True, batch_size=200)

    del datas_to_create
    gc.collect()
    
while True:
    do_from_schedule()
    time.sleep(5)

tracemalloc检测结果 #

内存泄漏持续增长，每次迭代的泄漏点均指向同一位置：

• #1: postgresql\operations.py:322: 156.9 KiB
  return cursor.query.decode()
• #1: postgresql\operations.py:322: 235.3 KiB
  return cursor.query.decode()
• #1: postgresql\operations.py:322: 313.7 KiB
  return cursor.query.decode()
• #1: postgresql\operations.py:322: 392.0 KiB
  return cursor.query.decode()
• #1: postgresql\operations.py:322: 470.4 KiB
  return cursor.query.decode()

可能的原因与修复方案 #

1. 显式管理数据库连接 #

PostgreSQL的psycopg2驱动在批量操作后，可能存在游标或查询缓存未被清理的情况。尝试手动关闭并重建连接，避免连接对象长期持有内存引用：

from django.db import connection

def do_from_schedule():
    # ... 原有代码 ...

    DataObject.objects.bulk_create(datas_to_create, ignore_conflicts=True, batch_size=200)

    # 手动关闭并重建连接
    connection.close()
    connection.connect()

    del datas_to_create
    gc.collect()

2. 缩小批量操作的批次大小 #

原batch_size=200可能过大，导致单次生成的SQL查询字符串（包含所有插入数据）被长期引用。尝试缩小批次，减少单次内存占用：

DataObject.objects.bulk_create(datas_to_create, ignore_conflicts=True, batch_size=50)

3. 优化模型实例的内存使用 #

避免长期持有大量模型实例引用，可改用字典构造插入数据，减少实例带来的额外内存开销：

datas_to_create = []
for load in loads:
    raw_data = json.dumps(load)
    name = json.loads(raw_data)['name']
    datas_to_create.append({
        'raw_data': raw_data,
        'name': name
    })
# 转换为模型实例后批量插入
DataObject.objects.bulk_create([DataObject(**item) for item in datas_to_create], ignore_conflicts=True, batch_size=50)

4. 升级psycopg2版本 #

旧版本的psycopg2可能存在已知的内存泄漏问题，升级到最新稳定版：

pip install --upgrade psycopg2-binary

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