如何用Python多线程优化百万行CSV文件解析的执行效率？

你的问题核心在于Python的GIL限制和任务类型的匹配，能不能用多线程提速，得看ParserFunction的具体逻辑：

一、先明确：多线程不一定适合所有场景 #

1. 如果ParserFunction是CPU密集型（比如大量计算、复杂字符串处理） #

Python的全局解释器锁（GIL）会限制同一时间只有一个线程执行Python字节码，多线程不仅不会提速，反而会因为线程切换增加额外开销，导致更慢。这时候应该用多进程，因为每个进程有独立的GIL，能真正利用多核CPU。

2. 如果ParserFunction是IO密集型（比如写数据库、调用API、读写其他文件） #

多线程能有效利用IO等待的空闲时间，让其他线程继续执行，这时候能明显缩短总耗时。

二、具体优化实现 #

场景1：IO密集型任务（用多线程） #

用concurrent.futures.ThreadPoolExecutor批量提交任务，避免一行一行提交的开销：

import csv
from concurrent.futures import ThreadPoolExecutor

def ParserFunction(row):
    # 你的IO密集型逻辑，比如写数据库、调用API等
    pass

def main():
    with open('csvfeed.csv', newline='', encoding='utf-8') as csvfile:
        reader = csv.reader(csvfile, delimiter=';', quotechar='|')
        # 内存充足的话直接读取所有行（100万行CSV内存压力通常不大）
        rows = list(reader)
        # 线程数根据IO场景调整，一般设10-20即可
        with ThreadPoolExecutor(max_workers=10) as executor:
            executor.map(ParserFunction, rows)

if __name__ == "__main__":
    main()

如果内存不足，可分批读取处理：

import csv
from concurrent.futures import ThreadPoolExecutor

def ParserFunction(row):
    # 你的IO密集型逻辑
    pass

def process_batch(batch):
    with ThreadPoolExecutor(max_workers=10) as executor:
        executor.map(ParserFunction, batch)

def main():
    batch_size = 1000  # 每批处理1000行
    batch = []
    with open('csvfeed.csv', newline='', encoding='utf-8') as csvfile:
        reader = csv.reader(csvfile, delimiter=';', quotechar='|')
        for row in reader:
            batch.append(row)
            if len(batch) >= batch_size:
                process_batch(batch)
                batch = []
        # 处理剩余的行
        if batch:
            process_batch(batch)

if __name__ == "__main__":
    main()

场景2：CPU密集型任务（用多进程） #

用concurrent.futures.ProcessPoolExecutor绕开GIL，利用多核CPU：

import csv
from concurrent.futures import ProcessPoolExecutor

def ParserFunction(row):
    # 你的CPU密集型逻辑，比如复杂计算、正则处理等
    pass

def main():
    with open('csvfeed.csv', newline='', encoding='utf-8') as csvfile:
        reader = csv.reader(csvfile, delimiter=';', quotechar='|')
        rows = list(reader)
        # 进程数默认设为CPU核心数，也可手动指定
        with ProcessPoolExecutor() as executor:
            executor.map(ParserFunction, rows)

if __name__ == "__main__":
    main()

内存不足时的分批处理逻辑和多线程版本类似，只需把ThreadPoolExecutor替换为ProcessPoolExecutor即可。

三、额外优化建议 #

1. 先定位瓶颈：用cProfile分析脚本，确认耗时到底在CSV读取还是ParserFunction，命令如下：

   python -m cProfile -s cumulative your_script.py
   

   如果是CSV读取慢，可以换成更快的第三方库（比如pandas适合结构化数据），或调整csv.reader的参数优化解析效率。

2. 减少数据传递开销：多进程之间传递数据会有序列化开销，尽量让每个进程处理一批数据而非单个行，同时让ParserFunction尽量是纯函数（仅依赖输入参数，不修改外部状态）。

3. 避免竞态条件：多线程/多进程环境下，全局变量容易引发数据混乱，尽量将状态封装在函数内部。

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