基于多线程的Django CSV数据迁移报错排查求助

嘿，我之前也踩过多线程导入Django数据库的坑，结合你说的加了异常捕获和锁还是报错的情况，咱们从几个常见的问题点入手来解决：

1. Django数据库连接的线程安全坑 #

Django的数据库连接是和线程绑定的，如果多个线程复用同一个连接，很容易出现连接重置、事务混乱这类奇怪的报错。解决办法很直接：让每个线程拥有独立的数据库连接，操作完主动关闭：

from django.db import connection

def process_chunk(chunk):
    # 每个线程启动时手动建立独立连接
    connection.connect()
    try:
        # 这里写你的数据转换和保存逻辑
        for row in chunk:
            # 比如把CSV行转成模型实例
            MyModel.objects.create(
                field1=row['col1'],
                field2=row['col2']
            )
    except Exception as e:
        print(f"Chunk处理失败: {str(e)}")
        connection.rollback()  # 出错回滚事务
    finally:
        # 线程结束前关闭连接，避免连接泄漏
        connection.close()

2. 锁机制用错了地方 #

如果你只加了Python的threading.Lock()，那只能保证代码层面的互斥，但数据库层面的并发冲突（比如唯一键重复、外键关联错误）还是会发生。给你两个更靠谱的方案：

• 针对唯一约束字段：用get_or_create()替代create()，避免重复插入报错：

  MyModel.objects.get_or_create(
      unique_field=row['unique_col'],
      defaults={'field1': row['col1'], 'field2': row['col2']}
  )
  

• 数据库层面原子操作：用transaction.atomic()包裹批量操作，要么全成功要么全失败，同时配合bulk_create提升效率：

  from django.db import transaction
  
  with transaction.atomic():
      # 批量创建，batch_size控制每次插入的数量，避免数据库压力过大
      MyModel.objects.bulk_create(
          [MyModel(**row) for row in chunk],
          batch_size=200
      )
  

3. 大型CSV分块的正确姿势 #

直接把整个CSV加载到内存再拆分，容易出现内存溢出或者线程间数据共享的问题。建议先把CSV拆成小的独立chunk，再分给不同线程处理：

import csv

def split_csv(file_path, chunk_size=1000):
    """把大型CSV拆分成多个小chunk"""
    with open(file_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        current_chunk = []
        for row in reader:
            current_chunk.append(row)
            if len(current_chunk) == chunk_size:
                yield current_chunk
                current_chunk = []
        # 处理最后不足chunk_size的部分
        if current_chunk:
            yield current_chunk

# 启动线程处理每个chunk
from threading import Thread

threads = []
for chunk in split_csv('your_large_file.csv'):
    t = Thread(target=process_chunk, args=(chunk,))
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

4. 异常捕获要精准 #

你说加了异常捕获，但可能没覆盖到Django特定的数据库异常。建议细化捕获逻辑，方便定位问题：

from django.db import IntegrityError, OperationalError

def process_chunk(chunk):
    connection.connect()
    try:
        with transaction.atomic():
            MyModel.objects.bulk_create([MyModel(**row) for row in chunk])
    except IntegrityError as e:
        # 处理唯一键、外键这类完整性错误
        print(f"数据完整性错误: {str(e)}，开始逐行排查")
        # 逐行处理找出错误数据
        for row in chunk:
            try:
                MyModel.objects.create(**row)
            except IntegrityError:
                print(f"错误行数据: {row}")
    except OperationalError as e:
        # 处理数据库连接超时、断开这类问题，可以加重试逻辑
        print(f"数据库连接错误: {str(e)}，重试当前chunk")
        process_chunk(chunk)
    except Exception as e:
        print(f"未知错误: {str(e)}")
    finally:
        connection.close()

5. 实在搞不定？试试现成工具 #

如果多线程的坑实在绕不过，不如用现成的解决方案：比如django-import-export这个第三方库，它已经帮你处理了线程安全、批量优化和错误处理的问题：

from import_export import resources
from .models import MyModel

class MyModelResource(resources.ModelResource):
    class Meta:
        model = MyModel

# 导入CSV，raise_errors=False会跳过错误行继续执行
resource = MyModelResource()
with open('your_large_file.csv', 'r', encoding='utf-8') as f:
    resource.import_data(f, dry_run=False, raise_errors=False)

最后提醒下：如果能把堆栈跟踪里的具体错误信息贴出来，比如是IntegrityError还是OperationalError，能更精准地定位问题哦！

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