如何正确实现 Excel 读取 → 解析 → 写入的流水线式并发处理

本文详解如何使用 ThreadPoolExecutor 和 ProcessPoolExecutor 构建非阻塞、阶段解耦的并发流水线，避免原代码中因 as_completed 顺序等待导致的串行瓶颈，并确保 I/O 与 CPU 密集型任务各司其职、无缝衔接。

本文详解如何使用 `threadpoolexecutor` 和 `processpoolexecutor` 构建非阻塞、阶段解耦的并发流水线，避免原代码中因 `as_completed` 顺序等待导致的串行瓶颈，并确保 i/o 与 cpu 密集型任务各司其职、无缝衔接。

在 Python 并发编程中，将 Excel 读取（I/O 密集）、DataFrame 解析（CPU 密集）和数据库写入（I/O 密集）三个阶段串联执行时，关键目标是实现“一个任务完成即触发下游处理”，而非等全部上游任务结束才批量推进。原始代码存在两个核心问题：

严重阻塞流水线：for tfuture in as_completed(tfutures) 必须等待所有 read_excel 任务完成才开始提交 parse_df；同理，parse_df 全部完成才启动 update_data。这实质上退化为三阶段串行批处理，完全丧失流水线并发价值；
资源管理不安全：手动调用 tpool.shutdown() 前未 wait=True，且未使用 with 语句，存在任务被强制终止或异常泄漏风险。

✅ 正确解法是：以线程池为调度中枢，每个线程内完成“读 → 解析 → 写”全链路，其中 CPU 密集的解析环节交由进程池异步执行并同步等待结果。这样既复用线程池轻量级调度能力，又让 CPU 工作在独立进程中并行，同时天然支持“单文件粒度流水线”。

以下为推荐实现（含完整可运行示例）：

Python对Excel操作详解中文WORD版

本文档主要介绍如何通过python对office excel进行读写操作，使用了xlrd、xlwt和xlutils模块。另外还演示了如何通过Tcl tcom包对excel操作。感兴趣的朋友可以过来看看

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import os

class ExcelProcessor:
    def __init__(self, config):
        self.config = config

    def worker(self, path, file, ppool):
        """单文件全流程处理：线程内执行，解析委托给进程池"""
        # 阶段1：I/O密集 —— 读取Excel（线程内直接调用）
        df_data = self._read_excel(path, file)

        # 阶段2：CPU密集 —— 解析DataFrame（交由进程池并行处理）
        parse_future = ppool.submit(self._parse_df, df_data)
        parsed_result = parse_future.result()  # 同步等待本文件解析完成（不阻塞其他线程）

        # 阶段3：I/O密集 —— 写入SQL（线程内直接调用）
        self.update_data(parsed_result[1], parsed_result[0])

    def read_excel(self):
        """主入口：启动线程池 + 进程池，按文件粒度分发任务"""
        file_list = ['分诊咨询记录', '客户消费明细表', '电话网络情况明细表']

        with ThreadPoolExecutor(max_workers=3) as tpool, \
             ProcessPoolExecutor(max_workers=3) as ppool:

            futures = []
            for file in file_list:
                path = os.path.join(self.config['his']['root'], file)
                if os.path.exists(path):
                    # 每个文件启动一个线程任务，worker内部完成三级流水线
                    future = tpool.submit(self.worker, path, file, ppool)
                    futures.append(future)

            # 等待所有文件处理完毕（隐式shutdown，安全可靠）
            for future in futures:
                future.result()  # 可选：捕获各文件处理异常

    # 以下为模拟方法（实际需替换为真实逻辑）
    def _read_excel(self, path, file):
        print(f"[READ] Processing {file} from {path}")
        return f"raw_{file}"  # 模拟返回DataFrame或数据结构

    def _parse_df(self, raw_data):
        print(f"[PARSE] CPU-intensive work on {raw_data}")
        # 模拟CPU密集操作（如复杂计算、清洗）
        return (raw_data + "_parsed", raw_data + "_meta")

    def update_data(self, data, meta):
        print(f"[WRITE] Saving to DB: data={data}, meta={meta}")

# 使用示例（Windows下必须加if __name__ == '__main__'）
if __name__ == '__main__':
    config = {'his': {'root': './data'}}
    processor = ExcelProcessor(config)
    processor.read_excel()

? 关键设计说明：

✅ 真流水线：当“分诊咨询记录”读取完成，其解析立即提交至进程池；该解析完成后立刻写入，全程不等待“客户消费明细表”的读取结果；
✅ 职责分明：ThreadPoolExecutor 调度 I/O 任务（读/写），ProcessPoolExecutor 承担 CPU 计算（解析），避免 GIL 限制；
✅ 安全可靠：with 语句确保池资源自动释放，future.result() 提供异常传播，便于错误定位；
⚠️ 注意事项：
- _parse_df 必须是纯函数（无共享状态、无副作用），因进程间不共享内存；
- 若解析逻辑依赖大对象（如模型、配置），需通过参数传入或使用 concurrent.futures.ProcessPoolExecutor 的 initializer；
- 对于极小文件或轻量解析，多进程开销可能超过收益，建议先性能测试，必要时统一用线程池。