如何使用 Kivy 在 Python 中进行多处理

问：

我不明白如何将多处理与 Python GUI 库 Kivy 相结合，所以请帮助我。

最初，我使用 Kivy、Python GUI 库和线程创建了一个代码。线程，如下所示。

#-*- coding: utf-8 -*-

from kivy.lang import Builder
Builder.load_string("""
<TextWidget>:
    BoxLayout:
        orientation: 'vertical'
        size: root.size

        Button:
            id: button1
            text: "start"
            font_size: 48
            on_press: root.buttonClicked()
""")

from kivy.app import App
from kivy.uix.widget import Widget

from kivy.properties import StringProperty 
import threading
import time

class TextWidget(Widget):
    def __init__(self, **kwargs):
        super(TextWidget, self).__init__(**kwargs)
        self.process_test = None
        
    def p_test(self):
        i = 0
        while True:
            print(i)
            i = i + 1
            
            if self.ids.button1.text == "start":
                break

    def buttonClicked(self):
        if self.ids.button1.text == "start":
            self.process_test = threading.Thread(target=self.p_test)
            self.process_test.start()
            self.ids.button1.text = "stop"
        else:
            self.ids.button1.text = "start"
            self.process_test.join()
            

class TestApp(App):
    def __init__(self, **kwargs):
        super(TestApp, self).__init__(**kwargs)
        self.title = 'testApp'

    def build(self):
        return TextWidget()

if __name__ == '__main__':
    TestApp().run()

此代码仅显示一个按钮，当按下该按钮时，它会在 while 循环中执行 print 语句。

此代码是一个简单的示例，可以正常工作。

然而，随着 Kivy GUI 定义文件变大，以及在 p_test 函数内运行的程序的 CPU 处理负载增加，程序开始变得不稳定。

根据我机器的任务管理器，尽管有足够的 CPU 容量，但在单个进程中处理所有内容似乎存在局限性。

为了规避这个问题，我决定使用多处理。但是，我发现多处理的使用很复杂且难以理解，因此我想了解更多关于如何使用它的信息。

首先，我更换了螺纹。在我之前的代码中使用多处理线程。流程如下。

#-*- coding: utf-8 -*-

from kivy.lang import Builder
Builder.load_string("""
<TextWidget>:
    BoxLayout:
        orientation: 'vertical'
        size: root.size

        Button:
            id: button1
            text: "start"
            font_size: 48
            on_press: root.buttonClicked()
""")

from kivy.app import App
from kivy.uix.widget import Widget

from kivy.properties import StringProperty 
import time
from multiprocessing import Process

class TextWidget(Widget):
    def __init__(self, **kwargs):
        super(TextWidget, self).__init__(**kwargs)
        self.process_test = None
        
    def p_test(self):
        i = 0
        while True:
            print(i)
            i = i + 1
            
            if self.ids.button1.text == "start":
                break

    def buttonClicked(self):
        if self.ids.button1.text == "start":
            self.process_test = Process(target=self.p_test, args=())
            self.process_test.start()
            self.ids.button1.text = "stop"
        else:
            self.ids.button1.text = "start"
            self.process_test.join()
            

class TestApp(App):
    def __init__(self, **kwargs):
        super(TestApp, self).__init__(**kwargs)
        self.title = 'testApp'

    def build(self):
        return TextWidget()

if __name__ == '__main__':
    TestApp().run()

不幸的是，此代码无法正常工作并导致错误。错误消息如下。

 Traceback (most recent call last):
   File "<string>", line 1, in <module>
   File "C:\Users\taichi\Documents\Winpython64-3.11.5.0\WPy64-31150\python-3.11.5.amd64\Lib\multiprocessing\spawn.py", line 122, in spawn_main
     exitcode = _main(fd, parent_sentinel)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "C:\Users\taichi\Documents\Winpython64-3.11.5.0\WPy64-31150\python-3.11.5.amd64\Lib\multiprocessing\spawn.py", line 132, in _main
     self = reduction.pickle.load(from_parent)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 EOFError: Ran out of input

我了解到我必须直接在 下使用多处理。但是，这使得在 Kivy 和多处理代码之间传递值变得困难，我不确定如何处理它。if **name** == '**main** ':

我作为试用版创建的代码如下。

#-*- coding: utf-8 -*-

from kivy.lang import Builder
Builder.load_string("""
<TextWidget>:
    BoxLayout:
        orientation: 'vertical'
        size: root.size

        Button:
            id: button1
            text: "start"
            font_size: 48
            on_press: root.buttonClicked()
""")

from kivy.app import App
from kivy.uix.widget import Widget

from kivy.properties import StringProperty 
import threading
import time
from multiprocessing import Process

class TextWidget(Widget):
    def __init__(self, **kwargs):
        super(TextWidget, self).__init__(**kwargs)
        self.process_test = None
                
    def buttonClicked(self):
        if self.ids.button1.text == "start":
            self.ids.button1.text = "stop"
        else:
            self.ids.button1.text = "start"
            

class TestApp(App):
    def __init__(self, **kwargs):
        super(TestApp, self).__init__(**kwargs)
        self.title = 'testApp'

    def build(self):
        return TextWidget()

def p_test(count, array):
    i = 0
    while True:
        print(i)
        i = i + 1

if __name__ == '__main__':
    #shared memory
    count = Value('i', 0)
    array = Array('i', 0)

    process_kivy = Process(target=TestApp().run(), args=[count, array])
    process_kivy.start()
    
    process_test = Process(target=p_test(), args=[count, array])
    process_test.start()
    
    process_kivy.join()
    process_test.join()

我之所以创建上述代码，是因为我了解到使用共享内存允许在多处理实例之间共享数据。但是，我不明白如何将数据传递给具有共享内存的类。

我想设置它，以便 while 循环仅在按下 Kivy 按钮时启动，但实际上，print 语句在此代码中关闭 Kivy GUI 后执行。

此外，由于 Kivy 程序也需要通过多处理启动，因此我不知道如何将我自己的进程加入到自身中。

如何正确使用多处理？

我正在使用 Windows11 和 WinPython。