Python多进程中如何让Function2等待Function1创建CSV文件后启动

针对你用ProcessPool并行调用Function1和Function2时，Function1创建CSV文件慢导致Function2报错的问题，以下是几种比固定time.sleep(5)更合理的解决方案：

方案一：使用multiprocessing.Event做进程同步 #

利用进程间的事件通知机制，让Function1创建完文件后触发事件，Function2等待事件触发后再读取文件。由于ProcessPool的子进程无法直接共享普通Event，需要通过Manager创建可跨进程共享的Event对象。

from multiprocessing import Pool, Manager

def Function1(event):
    # 创建CSV文件
    makecsv()
    # 通知Function2可以开始读取
    event.set()
    # 执行其他操作
    DoOtherStuff()
    return

def Function2(event):
    # 先执行自身前置操作
    DoStuff()
    # 等待Function1的启动通知
    event.wait()
    # 读取CSV文件
    readcsv()
    # 执行其他操作
    DoOtherStuff()
    return

if __name__ == "__main__":
    with Manager() as manager:
        # 创建可跨进程共享的Event
        start_event = manager.Event()
        with Pool(processes=2) as pool:
            # 并行启动两个函数，传递共享Event
            pool.apply_async(Function1, args=(start_event,))
            pool.apply_async(Function2, args=(start_event,))
            # 等待所有进程执行完成
            pool.close()
            pool.join()

方案二：轮询检查文件是否存在 #

如果不想依赖进程同步原语，可以让Function2循环检查目标CSV文件的存在性，一旦检测到文件创建完成就立即开始读取。这种方式无需额外同步对象，适合简单场景。

import os
import time
from multiprocessing import Pool

def Function1():
    makecsv()
    DoOtherStuff()
    return

def Function2(csv_file_path):
    DoStuff()
    # 每隔0.5秒检查一次文件是否存在
    while not os.path.exists(csv_file_path):
        time.sleep(0.5)
    # 文件存在后执行读取操作
    readcsv()
    DoOtherStuff()
    return

if __name__ == "__main__":
    csv_path = "target_file.csv"
    with Pool(processes=2) as pool:
        pool.apply_async(Function1)
        pool.apply_async(Function2, args=(csv_path,))
        pool.close()
        pool.join()

方案三：使用multiprocessing.Queue传递信号 #

通过队列传递触发信号，Function1创建完文件后往队列中发送消息，Function2等待接收消息后再执行读取操作。这种方式还可扩展传递更多业务信息。

from multiprocessing import Pool, Manager

def Function1(queue):
    makecsv()
    # 发送文件就绪信号
    queue.put("csv_ready")
    DoOtherStuff()
    return

def Function2(queue):
    DoStuff()
    # 等待队列中的启动信号
    queue.get()
    readcsv()
    DoOtherStuff()
    return

if __name__ == "__main__":
    with Manager() as manager:
        signal_queue = manager.Queue()
        with Pool(processes=2) as pool:
            pool.apply_async(Function1, args=(signal_queue,))
            pool.apply_async(Function2, args=(signal_queue,))
            pool.close()
            pool.join()

三种方案中，方案一的事件通知是进程同步的标准做法，逻辑清晰且效率最高；方案二适合对同步精度要求不高的简单场景；方案三的队列方式灵活性强，可适配更复杂的交互需求。

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