如何将大型multiprocessing.managers.ListProxy传递给subprocess.Popen？

1. ListProxy无法直接传给subprocess.Popen #

multiprocessing.managers.ListProxy是multiprocessing框架专属的代理对象，仅能在同一multiprocessing上下文的进程间共享，无法直接传递给subprocess.Popen创建的独立系统进程——两者属于完全不同的进程通信机制，直接传递会触发序列化错误。

2. 不能用json处理ListProxy #

• json不支持原生序列化numpy ndarray，必须先转换为普通列表或专用格式，否则会报错。
• ListProxy不是原生Python列表，是代理对象，直接传入json.dump会因为无法序列化代理结构而失败。

3. 传递大型数据给subprocess的高效方案 #

针对700MB的numpy数组，推荐以下低拷贝、高效的方案：

方案一：管道传递二进制numpy格式 #

父进程先拼接ListProxy中的数组，通过stdin管道以npy二进制格式传递给子进程，避免磁盘临时文件：

# 父进程代码
import subprocess
import numpy as np
from multiprocessing.managers import ListProxy

# 假设list_proxy是你的目标ListProxy
list_proxy: ListProxy = ...
big_array = np.concatenate(list(list_proxy))

# 启动非阻塞子进程
proc = subprocess.Popen(
    ["python", "save_worker.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

# 写入二进制数据后关闭stdin，父进程可继续执行其他任务
np.save(proc.stdin, big_array)
proc.stdin.close()

# 后续按需检查子进程状态，无需立即阻塞
# proc.wait()

# save_worker.py 子进程代码
import numpy as np
import sys

# 从stdin读取npy格式数据
big_array = np.load(sys.stdin.buffer)
# 保存到目标文件
np.save("final_data.npy", big_array)

方案二：临时文件传递路径 #

父进程先将拼接后的数组写入临时文件，把文件路径作为命令行参数传给子进程，实现无内存拷贝的传递：

# 父进程代码
import subprocess
import numpy as np
import tempfile
import os

list_proxy: ListProxy = ...
big_array = np.concatenate(list(list_proxy))

# 创建不自动删除的临时npy文件
with tempfile.NamedTemporaryFile(delete=False, suffix=".npy") as temp_f:
    np.save(temp_f, big_array)
    temp_path = temp_f.name

# 启动子进程，传递临时文件路径
proc = subprocess.Popen(["python", "save_worker.py", temp_path])

# 父进程继续执行其他逻辑，后续可清理临时文件
# proc.wait()
# os.unlink(temp_path)

# save_worker.py 子进程代码
import numpy as np
import sys

temp_path = sys.argv[1]
big_array = np.load(temp_path)
np.save("final_data.npy", big_array)

方案三：共享内存传递（零拷贝最优解） #

利用multiprocessing.shared_memory创建共享内存块，父进程写入数组后，将共享内存名称传给子进程，实现完全零拷贝的数据共享：

# 父进程代码
import subprocess
import numpy as np
from multiprocessing import shared_memory

list_proxy: ListProxy = ...
big_array = np.concatenate(list(list_proxy))

# 创建共享内存，大小匹配数组字节数
shm = shared_memory.SharedMemory(create=True, size=big_array.nbytes)
# 将数组映射到共享内存
shm_array = np.ndarray(big_array.shape, dtype=big_array.dtype, buffer=shm.buf)
shm_array[:] = big_array[:]

# 传递共享内存名称、数组形状和 dtype 给子进程
proc = subprocess.Popen(
    [
        "python", "save_worker.py",
        shm.name,
        str(big_array.shape),
        str(big_array.dtype)
    ]
)

# 父进程可继续执行，需等子进程读取完成后再销毁共享内存
# proc.wait()
# shm.close()
# shm.unlink()

# save_worker.py 子进程代码
import numpy as np
from multiprocessing import shared_memory
import sys

shm_name = sys.argv[1]
shape = tuple(map(int, sys.argv[2].strip('()').split(',')))
dtype = np.dtype(sys.argv[3])

# 连接到已创建的共享内存
shm = shared_memory.SharedMemory(name=shm_name)
# 映射为numpy数组
big_array = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
# 保存文件
np.save("final_data.npy", big_array)
# 关闭共享内存
shm.close()

4. subprocess解决阻塞问题的说明 #

subprocess.Popen本身是非阻塞调用，启动后父进程无需调用join()或wait()（除非主动需要等待子进程完成），可以直接继续执行其他任务，从而避免原multiprocessing子进程阻塞父进程的问题。

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