如何在Python多进程中实现子进程完成通知主进程的回调？

你的现有代码是等所有子进程执行完毕后才统一获取结果，无法实现实时通知主进程更新GUI的需求。下面提供两种可行的实时接收方案，适配GUI场景：

方案1：使用multiprocessing.Queue实现结果实时推送 #

队列是多进程安全的通信组件，子进程计算完成后直接将结果写入队列，主进程通过定时检查队列的方式，一旦收到结果就立即更新GUI。

修改后的代码示例 #

import multiprocessing
from multiprocessing import Process, Queue
import numpy as np
# 以Tkinter为例做GUI框架示例
import tkinter as tk

def worker(name, array, result_queue):
    # 执行计算任务
    s = sum(array)
    # 将进程标识和结果一起传入队列
    result_queue.put((name, s))

def check_queue(root, result_queue, result_label):
    # 非阻塞检查队列是否有新结果
    while not result_queue.empty():
        proc_name, result = result_queue.get()
        # 更新GUI组件，追加结果显示
        result_label.config(text=f"{result_label.cget('text')}\n{proc_name} 子进程结果: {result:.4f}")
    # 每隔100ms再次检查队列，避免阻塞GUI主循环
    root.after(100, check_queue, root, result_queue, result_label)

if __name__ == '__main__':
    # 初始化GUI
    root = tk.Tk()
    root.title("多进程实时结果展示")
    result_label = tk.Label(root, text="等待子进程结果...", justify=tk.LEFT)
    result_label.pack(padx=20, pady=20)

    # 创建结果队列
    result_queue = Queue()
    process_list = []

    # 启动子进程
    for i in range(5):
        a = np.random.random(10)
        print(i, ' 正确结果: ', a.sum())
        p = Process(target=worker, args=(str(i), a, result_queue))
        p.start()
        process_list.append(p)

    # 启动队列检查任务，融入GUI主循环
    check_queue(root, result_queue, result_label)

    # 启动GUI主循环
    root.mainloop()

    # 所有进程结束后做清理
    for p in process_list:
        p.join()

方案2：使用Pipe结合非阻塞接收 #

如果坚持使用Pipe，可以给每个子进程分配一对Pipe，主进程定期遍历所有Pipe的接收端，用poll()方法非阻塞检查是否有数据到达，有则读取结果并更新GUI。

修改后的代码示例 #

import multiprocessing
from multiprocessing import Process, Pipe
import numpy as np
import tkinter as tk

class MyProcess(Process):
    def __init__(self, name, array):
        super().__init__()
        self.name = name
        self.array = array
        self.recv_end, self.send_end = Pipe(False)

    def run(self):
        s = sum(self.array)
        self.send_end.send(s)

def check_pipes(root, process_list, result_label):
    for p in process_list:
        # 非阻塞检查Pipe是否有数据
        if p.recv_end.poll():
            result = p.recv_end.recv()
            result_label.config(text=f"{result_label.cget('text')}\n{p.name} 子进程结果: {result:.4f}")
    # 每隔100ms再次检查Pipe状态
    root.after(100, check_pipes, root, process_list, result_label)

if __name__ == '__main__':
    # 初始化GUI
    root = tk.Tk()
    root.title("多进程实时结果展示")
    result_label = tk.Label(root, text="等待子进程结果...", justify=tk.LEFT)
    result_label.pack(padx=20, pady=20)

    process_list = []
    for i in range(5):
        a = np.random.random(10)
        print(i, ' 正确结果: ', a.sum())
        p = MyProcess(str(i), a)
        p.start()
        process_list.append(p)

    # 启动Pipe检查任务
    check_pipes(root, process_list, result_label)

    root.mainloop()

    for p in process_list:
        p.join()

关键注意事项 #

1. GUI线程不能阻塞：无论使用哪种方案，都不能在GUI主循环中调用join()或阻塞式recv()，必须用非阻塞检查+定时回调的方式，或者将监听逻辑放在单独的线程（注意是线程而非进程，GUI组件只能在主线程更新）。
2. 通信安全：Queue和Pipe都是multiprocessing提供的多进程安全通信机制，避免直接共享内存引发的问题。
3. 结果标识：传递结果时最好带上进程名或唯一标识，方便在GUI中区分不同任务的计算结果。

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