Kivy结合多进程报错：进程池进程意外终止问题求助

问题背景 #

在Windows 10 + Python 3.10环境下，运行Kivy程序调用另一个使用ProcessPoolExecutor的文件处理模块时，间歇性出现错误："A process in the process pool was terminated abruptly while the future was running or pending"，程序有时能完整运行，有时中途崩溃。核心困惑是不知道如何在Kivy按钮调用场景下正确应用if __name__ == '__main__'保护机制。

原代码 #

proba.py（Kivy程序） #

# proba.py for kivy
from kivy.uix.widget import Widget
from kivy.app import App
import proba2

class Main(Widget):
    def __init__(self, **kwargs):
        super(Main, self).__init__(**kwargs)

    def Button(self):
        print ("W")
        proba2.File_read().Start()

class MainApp(App):
    def build(self):
        return Main()

if __name__ == '__main__':
    from kivy.lang import Builder
    Builder.load_string("""<Main>
    Button:
        on_press: root.Button()""")
    MainApp().run()

proba2.py（文件处理程序） #

# proba2.py for File_read
import concurrent.futures

class File_read():
    def __init__(self, **kwargs):
        super(File_read, self).__init__(**kwargs)

    def file_read (self, y, x):
        return y*x*self.Name  #Read files and give back to data

    def for_loop(self, Name):
        self.Name=Name
        results=[]
        results.clear()
        for_loop_result= []
        for_loop_result.clear()
        with concurrent.futures.ProcessPoolExecutor() as ex:
            for y in range (30):
                for x in range (30):
                    results.append (ex.submit(self.file_read,y,x))

            for f in concurrent.futures.as_completed(results):
                for_loop_result.append (f.result())
        return for_loop_result

    def Start(self):
        for Name_change in range (100):
            self.for_loop(Name_change)
            print ("Done")

问题根源 #

1. Windows多进程机制限制：Windows使用spawn方式创建子进程，会重新导入所有依赖模块。传递实例方法给ProcessPoolExecutor时，需要序列化整个类实例，极易引发不可预期的序列化/反序列化错误，导致进程异常终止。
2. Kivy主线程阻塞：直接在按钮回调中执行耗时的多进程任务，会阻塞Kivy界面渲染线程，不仅影响体验，还可能加剧进程调度冲突。

修复方案 #

修改proba2.py：改用独立函数传递参数 #

将依赖实例属性的方法改为独立函数，避免实例序列化问题：

# proba2.py for File_read
import concurrent.futures

# 改为独立函数，直接接收所需参数
def file_read(y, x, name):
    return y * x * name  # 实际场景替换为文件读取逻辑

class File_read():
    def __init__(self, **kwargs):
        super(File_read, self).__init__(**kwargs)

    def for_loop(self, name):
        results = []
        for_loop_result = []
        with concurrent.futures.ProcessPoolExecutor() as ex:
            for y in range(30):
                for x in range(30):
                    # 提交独立函数+参数，而非实例方法
                    results.append(ex.submit(file_read, y, x, name))

            for f in concurrent.futures.as_completed(results):
                for_loop_result.append(f.result())
        return for_loop_result

    def Start(self):
        for name_change in range(100):
            self.for_loop(name_change)
            print("Done")

修改proba.py：用线程包装耗时任务 #

避免阻塞Kivy主线程，保证界面响应：

# proba.py for kivy
from kivy.uix.widget import Widget
from kivy.app import App
import proba2
import threading

class Main(Widget):
    def __init__(self, **kwargs):
        super(Main, self).__init__(**kwargs)

    def Button(self):
        print("W")
        # 启动守护线程执行耗时操作，不阻塞Kivy主线程
        threading.Thread(target=proba2.File_read().Start, daemon=True).start()

class MainApp(App):
    def build(self):
        return Main()

if __name__ == '__main__':
    from kivy.lang import Builder
    Builder.load_string("""<Main>
    Button:
        on_press: root.Button()""")
    MainApp().run()

额外优化建议 #

1. 复用进程池：在Start方法中仅创建一次ProcessPoolExecutor，循环复用，减少进程创建销毁的开销。
2. 异常捕获：实际文件读取场景中，为file_read添加IO异常捕获，避免单个文件读取失败导致整个进程崩溃。
3. 控制进程数：根据系统资源设置ProcessPoolExecutor(max_workers=4)（建议为CPU核心数），避免资源耗尽。

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