本文介绍一种基于生产者-消费者模式的内存友好型并行文件处理方案：通过线程池执行 CPU 密集型行处理、独立写线程按序落盘，并借助队列限流避免内存溢出，兼顾顺序性、并发性与低内存占用。
本文介绍一种基于生产者-消费者模式的内存友好型并行文件处理方案：通过线程池执行 CPU 密集型行处理、独立写线程按序落盘，并借助队列限流避免内存溢出，兼顾顺序性、并发性与低内存占用。
在处理多 GB 级别文本文件（如日志、CSV）时，单纯依赖 readlines() 或 map() 会迅速耗尽内存；而盲目使用 ProcessPoolExecutor 或 ThreadPoolExecutor 又可能因 GIL 限制或进程开销导致性能不升反降。关键在于解耦「读取 → 计算 → 写入」三阶段，并显式控制执行顺序与资源边界。
✅ 核心设计：三线程协同 + 有序队列缓冲
本方案采用以下架构：
-
主线程：逐行流式读取输入文件（
for line in infile），零内存缓存； -
计算线程池（
ThreadPoolExecutor或ProcessPoolExecutor）：异步执行process_line()，返回Future对象； -
专用写线程：从有界队列中按提交顺序消费
Future，调用.result()阻塞等待完成后再写入输出文件，天然保证行序一致。
队列（queue.Queue(maxsize=...)）是关键——它既防止读取速度远超处理速度时内存爆炸（背压控制），又确保 Future 按提交顺序被消费，从而严格维持输入行序。
? 示例代码（可直接运行）
import concurrent.futures
import os
import queue
import threading
from io import IOBase
import time
def process_line(line: str) -> str:
# 替换为你的实际 CPU 密集型逻辑（如正则解析、数值计算）
# 注意：纯 I/O 或简单字符串操作（如 line.upper()）通常无需并行化
for _ in range(int(1e6)): # 模拟 CPU 负载
pass
return line.upper().rstrip('\n') + '\n'
def writer_task(out_file: IOBase, writer_queue: queue.Queue):
while True:
fut = writer_queue.get()
if fut is None: # 终止信号
break
try:
result = fut.result() # 阻塞直到该行处理完成
out_file.write(result)
except Exception as e:
# 生产环境必须捕获异常，否则写线程挂起将导致整个流程阻塞
print(f"Error writing line: {e}")
raise
finally:
writer_queue.task_done()
def main():
start_time = time.time()
with open("large_file.txt", "r") as infile, \
open("output.txt", "w") as outfile:
# 推荐：CPU 密集型任务优先试用 ProcessPoolExecutor（绕过 GIL）
# 若任务含较多 I/O 或全局解释器锁影响小，再对比 ThreadPoolExecutor
with concurrent.futures.ProcessPoolExecutor(
max_workers=os.cpu_count()
) as executor:
# 有界队列：容量 ≈ CPU 核心数 × 2 ~ 3，防内存失控
writer_queue = queue.Queue(maxsize=os.cpu_count() * 3)
writer = threading.Thread(
target=writer_task,
args=(outfile, writer_queue),
daemon=True # 避免主程序退出时残留
)
writer.start()
# 流式读取 + 提交任务（保持提交顺序）
for line in infile:
writer_queue.put(executor.submit(process_line, line))
# 发送终止信号并等待写入完成
writer_queue.put(None)
writer_queue.join() # 等待所有已入队任务写入完毕
end_time = time.time()
print(f"✅ 处理完成，耗时: {end_time - start_time:.2f} 秒")
if __name__ == "__main__":
main()
⚠️ 关键注意事项与最佳实践
-
Executor 选型需实测：
Secure Python Code Sandbox下载安全 Python 代码沙箱：在隔离的沙箱中执行任意 Python 代码，预装 requests、NumPy、pandas 库。返回 stdout、stderr。
-
ProcessPoolExecutor更适合纯 CPU-bound 任务（如加密、科学计算），但进程启动/序列化有开销； -
ThreadPoolExecutor适合 I/O-bound 或轻量计算，且无跨进程数据拷贝成本。
✅ 务必对真实负载进行基准测试（timeit/cProfile），勿凭经验猜测。
-
异常处理不可省略：
原始示例中fut.result()未捕获异常，一旦某行处理失败，写线程将永久阻塞。生产代码必须包裹try/except并合理传递错误（如记录日志、跳过该行或中止流程）。内存与性能平衡点：
queue.Queue(maxsize)的大小直接影响内存驻留行数。建议设为os.cpu_count() * 2 ~ 5—— 过小导致读取线程频繁等待，过大则削弱内存优势。避免微优化陷阱：
类似line.upper()这类极快操作，单线程处理反而更快（线程/进程调度开销 > 计算收益）。仅当单行处理耗时 ≥ 数毫秒时，并行才真正有益。-
进阶替代方案参考：
-
dask.bag：支持惰性求值与自动并行，适合复杂 ETL； -
pandarallel：针对 Pandas DataFrame 的并行加速； - 自定义分块 +
multiprocessing.Pool.imap_unordered+ 排序合并：适用于可接受分块内乱序、最终全局排序的场景。
-
该方案以最小依赖实现“流式读取 + 并行计算 + 有序写入”三位一体，在真实大数据预处理场景中已被验证稳定高效。记住：没有银弹，唯有测量驱动优化。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
