如何向Process/Thread传递共享列表？父进程列表为空问题求助

你遇到的问题核心，完全是Python多进程的内存隔离机制导致的，咱们一步步拆解清楚：

为什么父进程的列表始终为空？ #

Python的multiprocessing.Process创建子进程时，会触发「内存空间复制」（Unix系用fork机制，Windows会重新加载解释器并复刻父进程内存）。你传递给子进程的self.results列表，看起来是传引用，但跨进程时会被序列化/反序列化，最终在子进程里生成一个完全独立的副本。

也就是说：

• 子进程的_control函数拿到的results，和父进程的self.results是两个完全不同的对象，存在于不同的内存空间里
• 你在_receive线程里往这个副本里加元素，只会修改子进程自己的列表，父进程的原列表根本不会收到任何同步

你的代码还有个小bug要先修正 #

另外注意看你的_receive函数：

def _receive(results: list, pipe: IO):
    line = pipe.readline().decode('utf-8')
    while line is not '':
        results.append(results)  # 这里错了！
        line = pipe.readline().decode('utf-8')

你把列表自己添加到了列表里，这明显是笔误，应该改成results.append(line.strip())（或者你需要的处理后内容），不然就算共享成功，列表内容也会完全不符合预期。

解决办法：用进程安全的共享对象或通信机制 #

下面给你两种常用的可行方案：

方案1：用multiprocessing.Manager创建共享列表

Manager会生成一个跨进程共享的对象，所有进程操作的都是同一个实例，修改会实时同步：

1. 先导入Manager：

from multiprocessing import Process, Manager

2. 修改你的do_it方法：

def do_it(self):
    with Manager() as manager:
        self.results = manager.list()  # 创建跨进程共享的列表
        proc = Process(target=self._control, args=(self.results, self.path))
        proc.start()
        proc.join()
        print(self.results)  # 现在能看到子进程添加的结果了

剩下的_control和修正后的_receive函数不需要改动，直接用就行。

方案2：用multiprocessing.Queue传递结果

如果不需要实时共享列表，只是等子进程结束后收集结果，用队列更高效，也更符合进程间通信的常规模式：

1. 导入Queue：

from multiprocessing import Process, Queue

2. 修改do_it方法：

def do_it(self):
    result_queue = Queue()
    proc = Process(target=self._control, args=(result_queue, self.path))
    proc.start()
    proc.join()
    # 从队列取出所有结果，放到父进程的列表里
    self.results = []
    while not result_queue.empty():
        self.results.append(result_queue.get())
    print(self.results)

3. 同步修改_control和_receive函数：

def _control(self, result_queue, path):
    with Popen(path, stdout=PIPE, stderr=PIPE, stdin=PIPE) as proc:
        err_thread = threading.Thread(target=_read_errors, args=(proc,), daemon=True)
        err_thread.start()
        # 把队列传给接收线程
        receive_thread = threading.Thread(target=_receive, args=(result_queue, proc.stdout))
        receive_thread.start()
        # 这里发送你的指令...
        receive_thread.join()

def _receive(result_queue: Queue, pipe: IO):
    line = pipe.readline().decode('utf-8')
    while line != '':
        result_queue.put(line.strip())  # 把结果放入队列
        line = pipe.readline().decode('utf-8')

这种方式更安全，因为Queue本身就是线程和进程安全的，不会出现共享对象的竞争问题。

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