Ubuntu多进程最优通信方案及Python并行ML模型无文件数据共享咨询

嘿，针对你的问题，我来分享几个在Ubuntu+Python场景下完全替代临时文件的进程间通信方案，尤其适配你那三个并行运行的ML模型的需求：

在Ubuntu这类Linux系统中，针对不同场景有不同的最优选择，但结合你海量数据共享+跨终端进程同步的需求，优先考虑这两类：

• 共享内存：性能天花板，数据直接在进程间共享同一块内存区域，无需拷贝，特别适合海量数据场景。
• 基于IPC的消息框架：比如ZeroMQ，自带同步、重试机制，灵活性高，适合跨终端甚至跨机器的进程通信。

你的核心需求是：进程A生成批次数据后发就绪信号，进程B等信号再处理，全程不碰磁盘文件。下面是两个最适合的方案：

1. 共享内存+同步事件（性能最优） #

Python 3.8+自带的multiprocessing.shared_memory模块可以直接创建系统级共享内存，配合Event同步原语就能完美实现你的需求——数据存在内存里，不用写磁盘，信号同步精准。

举个贴合你场景的代码示例（假设用numpy存储ML批次数据）：

import multiprocessing as mp
import numpy as np
from multiprocessing import shared_memory

def model_a_process(shm_name, ready_event, batch_shape, dtype):
    """生成批次数据的进程A"""
    while True:
        # 生成一批ML数据（替换成你的数据生成逻辑）
        batch_data = np.random.rand(*batch_shape).astype(dtype)
        
        # 连接到已创建的共享内存
        existing_shm = shared_memory.SharedMemory(name=shm_name)
        # 将共享内存映射为numpy数组
        shared_batch = np.ndarray(batch_shape, dtype=dtype, buffer=existing_shm.buf)
        
        # 写入数据到共享内存
        shared_batch[:] = batch_data[:]
        # 发送数据就绪信号
        ready_event.set()
        
        # 等待进程B处理完当前批次，再生成下一批
        ready_event.wait()
        ready_event.clear()
        
        existing_shm.close()

def model_b_process(shm_name, ready_event, batch_shape, dtype):
    """处理数据的进程B"""
    # 连接到共享内存
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    shared_batch = np.ndarray(batch_shape, dtype=dtype, buffer=existing_shm.buf)
    
    while True:
        # 等待进程A的就绪信号
        ready_event.wait()
        # 执行ML模型处理（替换成你的模型推理逻辑）
        print(f"Processed batch - mean value: {shared_batch.mean():.4f}")
        
        # 通知进程A可以生成下一批数据
        ready_event.clear()
        ready_event.set()
    
    existing_shm.close()

if __name__ == "__main__":
    # 定义批次数据的规格（根据你的实际情况调整）
    BATCH_SHAPE = (2000, 100)  # 比如2000条样本，每条100特征
    DATA_TYPE = np.float32
    
    # 创建系统级共享内存，大小适配批次数据
    shm_size = np.prod(BATCH_SHAPE) * np.dtype(DATA_TYPE).itemsize
    shm = shared_memory.SharedMemory(create=True, size=shm_size)
    
    # 创建同步事件，用于传递就绪信号
    ready_event = mp.Event()
    
    # 启动两个进程（你可以在不同终端运行，只要共享内存名字和事件参数一致）
    # 注意：如果在不同终端运行，需要把进程启动逻辑拆分，比如单独写两个脚本分别启动A和B
    p_a = mp.Process(target=model_a_process, args=(shm.name, ready_event, BATCH_SHAPE, DATA_TYPE))
    p_b = mp.Process(target=model_b_process, args=(shm.name, ready_event, BATCH_SHAPE, DATA_TYPE))
    
    p_a.start()
    p_b.start()
    
    p_a.join()
    p_b.join()
    
    # 清理共享内存
    shm.close()
    shm.unlink()

方案优势： #

• 零磁盘IO，数据直接在内存中共享，性能拉满
• 同步逻辑清晰，事件信号精准控制批次流转
• 系统级共享内存，跨终端进程只要知道名字就能访问

2. ZeroMQ（IPC模式，灵活性高） #

如果你觉得共享内存的同步逻辑需要自己写有点麻烦，可以用ZeroMQ——一个高性能的消息库，支持IPC（进程间通信）模式，自带消息队列和同步机制，不用自己处理锁和事件。

示例代码：

import zmq
import numpy as np

def model_a_producer():
    """进程A：生成数据并推送"""
    context = zmq.Context()
    socket = context.socket(zmq.PUSH)
    # 绑定IPC地址，跨终端进程可以通过这个地址连接
    socket.bind("ipc:///tmp/ml_model_pipeline")
    
    while True:
        # 生成批次数据
        batch_data = np.random.rand(2000, 100).astype(np.float32)
        # 序列化数据：先发送数据长度，再发送字节数据
        socket.send(np.int32(batch_data.size).tobytes(), flags=zmq.SNDMORE)
        socket.send(batch_data.tobytes())

def model_b_consumer():
    """进程B：接收数据并处理"""
    context = zmq.Context()
    socket = context.socket(zmq.PULL)
    socket.connect("ipc:///tmp/ml_model_pipeline")
    
    while True:
        # 接收数据长度
        size_bytes = socket.recv()
        data_size = np.frombuffer(size_bytes, dtype=np.int32)[0]
        # 接收数据字节并还原为numpy数组
        data_bytes = socket.recv()
        batch_data = np.frombuffer(data_bytes, dtype=np.float32).reshape(2000, 100)
        
        # 处理数据（替换成你的模型逻辑）
        print(f"Processed batch - max value: {batch_data.max():.4f}")

# 拆分到不同终端运行的话，分别执行model_a_producer()和model_b_consumer()
if __name__ == "__main__":
    # 比如在终端1运行这个：
    # model_a_producer()
    # 在终端2运行这个：
    # model_b_consumer()
    pass

方案优势： #

• 无需手动处理同步逻辑，ZeroMQ自动帮你维护消息队列
• 支持多种通信模式（推送/拉取、请求/响应等），适配更复杂的多进程协作
• 可以轻松扩展到跨机器通信，不用改太多代码

• 如果追求极致性能，优先选共享内存+同步事件，完全无数据拷贝，适合海量数据场景
• 如果追求开发效率和灵活性，选ZeroMQ（IPC模式），自带的同步机制能省很多代码
• 不建议用multiprocessing.Queue，因为它会序列化并拷贝数据，海量数据下性能开销大

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