如何在Python的Kivy GUI库中正确使用multiprocessing多进程？

针对你在Windows 11 + WinPython环境下遇到的问题，以下是分步解决思路和代码示例：

1. 解决EOFError的核心前提 #

Windows系统下multiprocessing默认用spawn模式创建进程，这要求所有可能被子进程执行的代码必须被包裹在if __name__ == '__main__':块内，否则子进程会重新执行整个脚本，导致Kivy初始化异常，抛出EOFError。

错误根源：你之前直接替换Thread为Process但没做好主入口隔离，子进程重复加载Kivy GUI代码引发冲突。

2. 进程间通信的正确方式 #

Kivy的控件对象无法被序列化，不能直接传递给子进程。必须用multiprocessing提供的Queue或Pipe实现数据交互，避免用共享内存（共享内存适合简单数值，复杂场景不如Queue直观）。

3. 避免GUI卡顿与进程时序问题 #

• 子进程负责CPU密集型任务，不要在子进程中操作Kivy GUI
• 用Kivy的Clock.schedule_interval轮询队列，在主进程（GUI进程）中更新界面，避免阻塞事件循环
• 进程启动要在GUI初始化完成后，且不要阻塞主进程

完整代码示例 #

from kivy.app import App
from kivy.uix.button import Button
from kivy.uix.boxlayout import BoxLayout
from kivy.clock import Clock
import multiprocessing as mp
import time

# CPU密集型任务函数，运行在子进程
def cpu_intensive_task(queue):
    count = 0
    while True:
        count += 1
        # 将结果放入队列，传递给主进程
        queue.put(f"当前计数: {count}")
        time.sleep(0.1)  # 模拟负载，可替换为实际任务
        # 检查是否需要终止进程
        if not queue.empty() and queue.get() == "STOP":
            break

class MyApp(App):
    def build(self):
        self.layout = BoxLayout(orientation='vertical')
        self.start_btn = Button(text="启动任务")
        self.start_btn.bind(on_press=self.start_process)
        self.status_label = Button(text="等待任务启动", disabled=True)
        self.layout.add_widget(self.start_btn)
        self.layout.add_widget(self.status_label)
        
        # 初始化进程通信队列
        self.queue = mp.Queue()
        self.process = None
        
        # 注册GUI关闭时的清理函数
        self.bind(on_stop=self.cleanup_process)
        
        # 启动队列轮询，每0.1秒检查一次
        Clock.schedule_interval(self.update_gui, 0.1)
        return self.layout

    def start_process(self, instance):
        if self.process is None or not self.process.is_alive():
            # 创建并启动子进程，传递队列
            self.process = mp.Process(target=cpu_intensive_task, args=(self.queue,))
            self.process.start()
            self.start_btn.text = "任务运行中"
            self.start_btn.disabled = True

    def update_gui(self, dt):
        # 从队列获取子进程数据，更新GUI
        while not self.queue.empty():
            msg = self.queue.get()
            self.status_label.text = msg

    def cleanup_process(self, instance):
        # GUI关闭时终止子进程并等待结束
        if self.process and self.process.is_alive():
            self.queue.put("STOP")
            self.process.join(timeout=1)
            if self.process.is_alive():
                self.process.terminate()

if __name__ == '__main__':
    # 所有Kivy和进程启动逻辑必须在主入口块内
    MyApp().run()

关键细节解释 #

• 主入口保护：if __name__ == '__main__':确保子进程不会重复初始化Kivy App，解决EOFError
• Queue通信：子进程只负责计算并发送数据，主进程通过轮询更新GUI，完全隔离CPU密集任务和GUI线程
• 进程清理：在App的on_stop事件中发送终止信号，先尝试优雅结束，超时则强制终止，避免僵尸进程
• Clock轮询：Kivy的事件循环是非阻塞的，用schedule_interval可以安全地在主进程中处理子进程数据，不会导致GUI卡顿

针对你遇到的其他问题的说明 #

• 无法向类传递数据：不要直接传递Kivy类实例给子进程，用Queue中转数据即可
• 循环在GUI关闭后才执行：之前的代码可能把进程启动放在了主入口块外，导致子进程被延迟启动，现在的代码确保进程在按钮点击时立即启动
• 进程join的处理：join要放在GUI关闭时执行，避免阻塞主进程的事件循环，同时设置超时防止无限等待

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