Python使用yield生成器时如何实现文件并行加载处理？

可行性结论 #

边迭代当前文件边预加载下一个文件的需求完全可实现，该方案可以将读取下一个文件的IO等待时间和当前文件的处理时间重叠，大幅提升整体处理效率。

现有代码的问题点 #

• 多进程内存隔离：Python多进程模式下各个进程拥有独立的内存空间，你在load子进程中修改全局cache变量，只会修改子进程自己的内存副本，主进程的cache完全不会更新，自然会抛出KeyError
• 生成器无法跨进程返回结果：你用子进程执行iter_list生成器，仅会在子进程中创建生成器对象，既不会实际迭代内容，也无法把结果传回主进程的调用方，完全达不到逐行返回的要求
• multiprocessing启动机制限制：Windows和macOS默认使用spawn模式启动多进程，该模式会重新导入整个主模块，若没有if __name__ == "__main__"防护，导入时会重复执行启动进程的代码，触发递归报错，这是multiprocessing的强制规则，无法绕开。

推荐实现方案 #

因为你的场景以文件读取这类IO操作为主，属于IO密集型任务，使用多线程即可满足性能要求，还可以规避多进程的各种限制，不需要if __name__ == "__main__"防护，也没有内存隔离的问题。

实现代码如下：

import threading
from queue import Queue

def load(f_name: str, result_queue: Queue):
    with open(f_name, "r", encoding="utf-8") as f:
        lines = f.readlines()
    # 保留你原有的计算逻辑
    # some calculations
    result_queue.put(lines)

def iter_files(f_names: list):
    if not f_names:
        return
    # 队列用于存储预加载的文件内容，最大长度1保证仅预加载下一个文件
    load_queue = Queue(maxsize=1)
    # 先预加载第一个文件
    first_thread = threading.Thread(
        target=load,
        args=(f_names[0], load_queue),
        daemon=True
    )
    first_thread.start()

    for idx, current_f in enumerate(f_names):
        # 取出当前已加载完成的文件内容
        current_lines = load_queue.get()
        # 启动下一个文件的预加载（如果存在）
        if idx + 1 < len(f_names):
            next_thread = threading.Thread(
                target=load,
                args=(f_names[idx+1], load_queue),
                daemon=True
            )
            next_thread.start()
        # 逐行返回当前文件内容
        for line in current_lines:
            yield line

方案说明 #

• 用线程安全的Queue传递加载结果，不需要全局变量，避免多线程并发修改问题
• 每次取当前文件内容后，立刻启动下一个文件的加载线程，加载下一个文件的IO时间和当前文件的迭代处理时间完全重叠，可大幅降低总耗时
• 守护线程设置避免程序退出时线程残留

如果你的some calculations逻辑是重CPU计算，多线程受GIL限制性能不足，必须使用多进程的话，可以将上述代码中的threading.Thread替换为multiprocessing.Process，queue.Queue替换为multiprocessing.Queue即可，但此时必须遵守multiprocessing的规则，将主调用逻辑放到if __name__ == "__main__"代码块下。

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