PyQt使用multiprocessing多进程时消息实时更新GUI的实现问题

你遇到的消息串行展示问题核心出在Relay线程的执行逻辑上：

def run(self):
    for liaison in self.liaisons:
        liaison.listen()

listen()方法里的pipe.recv()是阻塞调用，只要第一个子进程没有完成全量消息发送，整个Relay线程就会一直堵在第一个Liaison的接收逻辑里，根本不会处理后续子进程的Pipe，导致后面子进程的消息全部积压，等前一个子进程执行完退出循环后才会批量处理下一个的消息。

方案1：单线程轮询所有Pipe（推荐，轻量高效） #

不需要开多个线程，只用单个Relay线程配合select轮询所有Pipe的可读状态，哪个Pipe有数据就读哪个，完全避免阻塞：

import select

class Relay(QtCore.QThread):
    message = QtCore.pyqtSignal(str)
    kill = QtCore.pyqtSignal(int)

    def __init__(self, connections):
        super().__init__()
        self.conn_map = {pipe: idx for idx, pipe in enumerate(connections)}
        self.stop_flag = False

    def run(self):
        while not self.stop_flag and self.conn_map:
            # 轮询所有可读的Pipe，超时设为0.1秒避免卡死
            readable, _, _ = select.select(self.conn_map.keys(), [], [], 0.1)
            for conn in readable:
                try:
                    msg = conn.recv()
                    worker_id = self.conn_map[conn]
                    if msg == RelayMessages.stop:
                        self.message.emit(f'Liaison {worker_id} stopping listening')
                        self.kill.emit(worker_id)
                        del self.conn_map[conn]
                        continue
                    self.message.emit(str(msg))
                except EOFError:
                    del self.conn_map[conn]

这个方案可以直接删掉原来的Liaison类，只需要把Relay的message信号绑定到UI更新函数即可，代码更简洁，也不会出现阻塞问题。

方案2：每个Liaison绑定独立线程 #

如果希望保留原来的Liaison结构，可以给每个Liaison单独分配一个QThread，各个监听逻辑独立运行互不干扰：

class Relay(QtCore.QObject):
    def __init__(self, connections):
        super().__init__()
        self.threads = []
        self.liaisons = []

        for id, pipe in enumerate(connections):
            t = QtCore.QThread()
            liaison = Liaison(id, pipe)
            liaison.moveToThread(t)
            t.started.connect(liaison.listen)
            liaison.message.connect(self.forward_msg)
            liaison.kill.connect(self.forward_kill)
            self.liaisons.append(liaison)
            self.threads.append(t)

    def start(self):
        for t in self.threads:
            t.start()

1. 不要改用QThreadPool：QThreadPool是线程池实现，Python线程受GIL限制，完全不适合你的CPU密集型转换场景，多进程是正确选择。
2. 如果你想用multiprocessing.Pool简化代码，可以配合全局mp.Queue实现子进程和主进程的通信：
   • 初始化一个全局的消息队列，所有子进程处理过程中把状态消息推入队列
   • 主进程开一个单独的QThread循环监听队列，收到消息就发信号更新UI
   • 直接用apply_async提交任务即可，不需要自己管理子进程生命周期和Pipe，代码会简洁很多。
     示例代码片段：

import multiprocessing as mp
msg_queue = mp.Queue()

# 子进程执行的任务函数
def worker_task(work_load, worker_id):
    msg_queue.put(f"worker ({worker_id}): task will take {work_load} seconds")
    for i in range(work_load):
        time.sleep(1)
        msg_queue.put(f"worker ({worker_id}): {i}")
    msg_queue.put(f"worker ({worker_id}): task complete")
    return work_load

# 主进程的队列监听线程
class QueueListener(QtCore.QThread):
    message = QtCore.pyqtSignal(str)
    def run(self):
        while True:
            try:
                msg = msg_queue.get()
                self.message.emit(msg)
            except:
                break

Windows系统下使用全局Queue需要注意把队列定义放在if __name__ == '__main__'之外，避免子进程导入模块时重复初始化。

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