concurrent.futures多线程/多进程脚本问题及并行诉求咨询

1. 你的脚本大概率存在问题 #

如果你的原始脚本是按「先批量提交所有读取任务→等全部读完→再批量提交解析任务→等全部解析完→再批量提交写入任务」的顺序写的，那肯定有问题：

• 会出现阶段式阻塞：必须等上一阶段所有任务结束才能进入下一阶段，完全浪费了异步处理的优势
• 资源利用率极低：比如读取是IO密集型任务，CPU会闲置；等全部读完再解析，又会让IO资源闲置，来回浪费

2. 原始脚本不符合需求，实现流水线异步处理的方法 #

你的核心需求是流水线式异步执行：一个文件读完就立刻解析，解析完就立刻写入，不需要等其他文件的前置任务完成。要实现这个，得用任务回调链把三个阶段串起来，让每个任务完成后自动触发下一阶段的任务。

实现思路 #

• 任务分工匹配执行器：
  • 读取Excel（IO密集）→ 用ThreadPoolExecutor
  • 解析DataFrame（CPU密集，比如数据清洗、计算）→ 用ProcessPoolExecutor（避开GIL限制）
  • 写入SQL（IO密集）→ 用ThreadPoolExecutor
• 用add_done_callback给每个任务绑定回调：当前任务完成后，自动把结果传给下一个阶段的任务，并提交到对应的执行器

代码示例 #

import pandas as pd
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# 读取Excel（IO密集任务）
def read_excel(file_path):
    print(f"开始读取: {file_path}")
    df = pd.read_excel(file_path)
    print(f"读取完成: {file_path}")
    return (file_path, df)

# 解析DataFrame（CPU密集任务）
def parse_df(file_tuple):
    file_path, df = file_tuple
    print(f"开始解析: {file_path}")
    # 这里替换成你的实际解析逻辑：清洗、转换等
    df['processed_col'] = df['original_col'] * 2
    print(f"解析完成: {file_path}")
    return (file_path, df)

# 写入SQL（IO密集任务）
def write_to_sql(file_tuple):
    file_path, df = file_tuple
    print(f"开始写入: {file_path}")
    # 这里替换成你的实际写入逻辑：df.to_sql(...)
    print(f"写入完成: {file_path}")
    return f"{file_path} 全流程完成"

def main():
    # 待处理的Excel文件列表
    file_list = ["file1.xlsx", "file2.xlsx", "file3.xlsx"]

    # 初始化不同类型的执行器
    read_pool = ThreadPoolExecutor(max_workers=3)  # IO密集可多开线程
    parse_pool = ProcessPoolExecutor(max_workers=2)  # CPU密集按核心数设
    write_pool = ThreadPoolExecutor(max_workers=2)

    # 提交读取任务，并绑定解析回调
    for file_path in file_list:
        read_future = read_pool.submit(read_excel, file_path)
        
        # 读取完成后触发解析任务
        def trigger_parse(fut):
            try:
                read_result = fut.result()
                parse_future = parse_pool.submit(parse_df, read_result)
                
                # 解析完成后触发写入任务
                def trigger_write(parse_fut):
                    try:
                        parse_result = parse_fut.result()
                        write_pool.submit(write_to_sql, parse_result)
                    except Exception as e:
                        print(f"解析任务失败: {e}")
                parse_future.add_done_callback(trigger_write)
            except Exception as e:
                print(f"读取任务失败: {e}")
        
        read_future.add_done_callback(trigger_parse)

    # 等待所有执行器完成任务（主线程可选择不等待，根据业务需求）
    read_pool.shutdown(wait=True)
    parse_pool.shutdown(wait=True)
    write_pool.shutdown(wait=True)
    print("所有文件全流程处理完成")

if __name__ == "__main__":
    main()

关键优势 #

• 真正的流水线：第一个文件读取完成后，立刻启动解析，解析完立刻写入，全程无阶段阻塞
• 资源高效利用：IO密集用线程池、CPU密集用进程池，最大化机器性能
• 异常隔离：每个阶段的异常都被捕获，不会影响其他文件的处理

注意事项 #

• 进程池传递的参数必须可序列化（因为进程间用pickle通信），所以返回的tuple里不能包含不可序列化的对象
• 执行器的max_workers要根据机器配置调整：线程池可设为CPU核心数*2，进程池设为CPU核心数左右
• 如果写入SQL需要数据库连接，注意不要在进程/线程里重复创建大量连接，建议用连接池

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