使用ijson读取大JSON文件时多进程处理失效问题求助

问题分析 #

1. 全局列表无法跨进程共享：你定义的player列表是全局变量，每个子进程都会创建自己的副本，主进程的列表不会被修改，最终无法得到有效结果。
2. pool.map的使用限制：map会尝试先遍历完整个可迭代对象再分配任务，而ijson.items是流式生成器，主进程逐个生成item的速度可能远慢于worker的处理速度，导致大部分worker处于等待状态，看起来只有一个在干活。
3. 缺少合理的任务分块：默认的任务分块大小过小，进程间通信开销大，也会导致负载分配不均。

修正后的代码 #

import ijson
import multiprocessing

file = 'jsonfile'

def process_game_record(record):
    player_list = []
    for game01 in record["games"]:
        try:
            player_list.append(game01['player'])
        except KeyError:
            player_list.append('No_record_found')
    return player_list

if __name__ == '__main__':
    num_workers = multiprocessing.cpu_count()
    
    with open(file, "rb") as f:
        with multiprocessing.Pool(num_workers) as pool:
            # 使用imap_unordered流式分配任务，设置chunksize减少通信开销
            results = pool.imap_unordered(
                process_game_record, 
                ijson.items(f, "game.item"), 
                chunksize=100  # 可根据实际情况调整大小
            )
            
            # 汇总所有子进程返回的结果
            all_players = []
            for result in results:
                all_players.extend(result)
    
    print(f"共处理{len(all_players)}条玩家记录")

关键优化点 #

• 改用返回值收集结果：每个worker处理完单个record后返回对应的玩家列表，主进程迭代汇总结果，避免跨进程共享变量的问题。
• 流式任务分配：imap_unordered会在生成item后立即分配给空闲的worker，无需等待所有item生成，让多个worker持续工作。
• 调整chunksize：将多个item打包成一块分配给worker，减少进程间通信的次数，提升整体效率。可根据文件中record的大小和数量调整，比如100、500或1000。
• CPU核心数适配：用cpu_count()设置进程数，最大化利用硬件资源。

额外注意 #

如果ijson读取item的速度成为新的瓶颈，可考虑将大JSON文件分割为多个小文件，用多进程分别读取处理；或者尝试更高效的JSON流式解析工具。

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