如何将Python同步CSV处理代码转换为异步代码以优化性能？

原代码的核心性能瓶颈在于循环中频繁发起单条数据库查询/写入请求，大量IO等待导致耗时过长。先优化同步逻辑可大幅降低耗时，再结合异步改造进一步提升并发性能。

第一步：同步代码优化（优先解决核心耗时问题） #

原代码问题点 #

• 循环内重复查询相同的Owner/Lot对象，产生大量冗余数据库请求
• create方法调用后额外执行save()，属于冗余操作（create已完成持久化）
• 单条处理数据，未利用数据库批量操作能力

优化后的同步代码 #

import pandas as pd
from django.db.models import Q

def handle_csv(csv):
    database = pd.read_csv(csv)
    
    # 提取并去重数据，避免重复处理
    owners_data = database[['account_number', 'first_name', 'last_name']].drop_duplicates(subset='account_number')
    lots_data = database[['lot_number', 'lot_description']].drop_duplicates(subset='lot_number')
    lot_owner_mapping = database[['lot_number', 'account_number']].drop_duplicates()

    # 批量查询已存在的Owner，减少查询次数
    existing_accounts = set(Owner.objects.filter(
        account_number__in=owners_data['account_number'].tolist()
    ).values_list('account_number', flat=True))
    
    # 批量创建不存在的Owner
    owners_to_create = []
    for _, row in owners_data.iterrows():
        if row['account_number'] not in existing_accounts:
            owners_to_create.append(Owner(
                account_number=row['account_number'],
                first_name=row['first_name'],
                last_name=row['last_name']
            ))
    Owner.objects.bulk_create(owners_to_create)

    # 批量查询已存在的Lot
    existing_lots = set(Lot.objects.filter(
        lot_number__in=lots_data['lot_number'].tolist()
    ).values_list('lot_number', flat=True))
    
    # 批量创建不存在的Lot
    lots_to_create = []
    for _, row in lots_data.iterrows():
        if row['lot_number'] not in existing_lots:
            lots_to_create.append(Lot(
                lot_number=row['lot_number'],
                lot_description=row['lot_description']
            ))
    Lot.objects.bulk_create(lots_to_create)

    # 批量获取Owner和Lot的映射关系
    owner_map = {o.account_number: o for o in Owner.objects.filter(
        account_number__in=owners_data['account_number'].tolist()
    )}
    lot_map = {l.lot_number: l for l in Lot.objects.filter(
        lot_number__in=lots_data['lot_number'].tolist()
    )}

    # 批量关联Lot与Owner（用关联表bulk_create效率更高）
    from your_app.models import LotOwnerThrough  # 替换为你的实际关联模型类名
    associations = []
    for _, row in lot_owner_mapping.iterrows():
        associations.append(LotOwnerThrough(
            lot_id=lot_map[row['lot_number']].id,
            owner_id=owner_map[row['account_number']].id
        ))
    LotOwnerThrough.objects.bulk_create(associations)

第二步：异步改造（进一步提升IO并发性能） #

若你的项目基于Django 3.1+，可利用异步ORM实现数据库IO并发，减少等待时间：

异步优化代码 #

import pandas as pd
from django.db.models import Q

async def handle_csv_async(csv):
    # CSV读取为CPU操作，同步执行即可
    database = pd.read_csv(csv)
    
    owners_data = database[['account_number', 'first_name', 'last_name']].drop_duplicates(subset='account_number')
    lots_data = database[['lot_number', 'lot_description']].drop_duplicates(subset='lot_number')
    lot_owner_mapping = database[['lot_number', 'account_number']].drop_duplicates()

    # 异步批量查询已存在的Owner
    existing_accounts = set()
    async for account in Owner.objects.filter(
        account_number__in=owners_data['account_number'].tolist()
    ).values_list('account_number', flat=True):
        existing_accounts.add(account)
    
    # 异步批量创建Owner
    owners_to_create = []
    for _, row in owners_data.iterrows():
        if row['account_number'] not in existing_accounts:
            owners_to_create.append(Owner(
                account_number=row['account_number'],
                first_name=row['first_name'],
                last_name=row['last_name']
            ))
    if owners_to_create:
        await Owner.objects.abulk_create(owners_to_create)

    # 异步批量查询已存在的Lot
    existing_lots = set()
    async for lot_num in Lot.objects.filter(
        lot_number__in=lots_data['lot_number'].tolist()
    ).values_list('lot_number', flat=True):
        existing_lots.add(lot_num)
    
    # 异步批量创建Lot
    lots_to_create = []
    for _, row in lots_data.iterrows():
        if row['lot_number'] not in existing_lots:
            lots_to_create.append(Lot(
                lot_number=row['lot_number'],
                lot_description=row['lot_description']
            ))
    if lots_to_create:
        await Lot.objects.abulk_create(lots_to_create)

    # 异步批量获取Owner和Lot映射
    owner_map = {}
    async for owner in Owner.objects.filter(
        account_number__in=owners_data['account_number'].tolist()
    ):
        owner_map[owner.account_number] = owner
    
    lot_map = {}
    async for lot in Lot.objects.filter(
        lot_number__in=lots_data['lot_number'].tolist()
    ):
        lot_map[lot.lot_number] = lot

    # 异步批量关联Lot与Owner
    from your_app.models import LotOwnerThrough
    associations = []
    for _, row in lot_owner_mapping.iterrows():
        associations.append(LotOwnerThrough(
            lot_id=lot_map[row['lot_number']].id,
            owner_id=owner_map[row['account_number']].id
        ))
    if associations:
        await LotOwnerThrough.objects.abulk_create(associations)

关键优化点 #

• 减少数据库查询次数：先批量查询已存在的对象，避免循环内重复发起请求
• 批量操作替代单条处理：用bulk_create/abulk_create将多次数据库请求合并为一次，大幅降低IO开销
• 提前数据去重：对CSV中的Owner、Lot数据去重，避免重复创建或关联操作
• 异步IO并发：利用Django异步ORM实现数据库请求并发执行，减少等待时间
• 移除冗余操作：删除create后的save()调用，create方法已完成对象持久化

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