Python多进程字符串共享内存结合Kivy报错的解决方法

已成功实现的案例 #

1. 简单Python字符串共享内存 #

from multiprocessing import Value, Array, Process, Lock

def process1(array,  lock):
    with lock:
        buffer_data = str("z1111").encode("utf-8")
        array[0:len(buffer_data)] = buffer_data[:]
        
if __name__ == '__main__':
    array = Array('c', 35)
    lock = Lock()

    process_test1 = Process(target=process1, args=[array, lock], daemon=True)
    
    process_test1.start()
    process_test1.join()
    
    print(array[:].decode("utf-8"))
    print("process ended")

2. Kivy框架下数值共享内存 #

from multiprocessing import Process, Value, set_start_method, freeze_support, Lock

kvLoadString=("""
<TextWidget>:
    BoxLayout:
        orientation: 'vertical'
        size: root.size

        Button:
            id: button1
            text: "start"
            font_size: 48
            on_press: root.buttonClicked()
        Label:
            id: lab
            text: 'result'
""")

def start_process(count,lock):
    process_test = Process(target=p_test, args=[count,lock], daemon=True)
    process_test.start()
    return process_test

def p_test(count,lock):
    i = 0
    while True:
        print(i)
        i = i + 1
        with lock:
            count.value = i

if __name__ == '__main__':
    freeze_support() 
    from kivy.app import App
    from kivy.uix.widget import Widget
    from kivy.lang import Builder

    from kivy.uix.checkbox import CheckBox
    from kivy.uix.spinner import Spinner
    from kivy.properties import StringProperty,NumericProperty,ObjectProperty, BooleanProperty
    from kivy.core.window import Window
    from kivy.uix.boxlayout import BoxLayout
    from kivy.uix.popup import Popup
    from kivy.uix.treeview import TreeView, TreeViewLabel, TreeViewNode
    from kivy.uix.label import Label
    from kivy.uix.textinput import TextInput
    from kivy.uix.progressbar import ProgressBar
    from kivy.uix.button import Button

    set_start_method('spawn')


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

        def build(self):
            return TextWidget()

    class TextWidget(Widget):
        def __init__(self, **kwargs):
            super(TextWidget, self).__init__(**kwargs)
            self.process_test = None
            self.proc = None

            # shared memory
            self.count = Value('i', 0)
            self.lock = Lock()

        def buttonClicked(self):
            if self.ids.button1.text == "start":
                self.proc = start_process(self.count,self.lock)
                self.ids.button1.text = "stop"
            else:
                if self.proc:
                    self.proc.kill()
                self.ids.button1.text = "start"
                self.ids.lab.text = str(self.count.value)

    Builder.load_string(kvLoadString)
    TestApp().run()

失败的Kivy字符串共享内存实现 #

from multiprocessing import Array,Process, Value, set_start_method, freeze_support, Lock
import math

kvLoadString=("""
<TextWidget>:
    BoxLayout:
        orientation: 'vertical'
        size: root.size

        Button:
            id: button1
            text: "start"
            font_size: 48
            on_press: root.buttonClicked()
        Label:
            id: lab
            text: 'result'
""")

def start_process(array,lock):
    process_test = Process(target=p_test, args=[array,lock], daemon=True)
    process_test.start()
    return process_test

def p_test(array,lock):
    with lock:
        buffer_data = str("z1111").encode("utf-8")
        array[0:len(buffer_data)] = buffer_data[:]
        print(array.decode("utf-8"))

if __name__ == '__main__':
    freeze_support() 
    from kivy.app import App
    from kivy.uix.widget import Widget
    from kivy.lang import Builder

    from kivy.uix.checkbox import CheckBox
    from kivy.uix.spinner import Spinner
    from kivy.properties import StringProperty,NumericProperty,ObjectProperty, BooleanProperty
    from kivy.core.window import Window
    from kivy.uix.boxlayout import BoxLayout
    from kivy.uix.popup import Popup
    from kivy.uix.treeview import TreeView, TreeViewLabel, TreeViewNode
    from kivy.uix.label import Label
    from kivy.uix.textinput import TextInput
    from kivy.uix.progressbar import ProgressBar
    from kivy.uix.button import Button

    set_start_method('spawn')


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

        def build(self):
            return TextWidget()

    class TextWidget(Widget):
        def __init__(self, **kwargs):
            super(TextWidget, self).__init__(**kwargs)
            self.process_test = None
            self.proc = None

            # shared memory
            self.array = Array('c', 35)
            self.lock = Lock()

        def buttonClicked(self):
            if self.ids.button1.text == "start":
                self.proc = start_process(self.array,self.lock)
                self.ids.button1.text = "stop"
            else:
                if self.proc:
                    self.proc.kill()
                self.ids.button1.text = "start"
                self.ids.lab.text = self.array.decode("utf-8")

    Builder.load_string(kvLoadString)
    TestApp().run()

报错信息 #

Traceback (most recent call last):
  File "C:\Users\taichi\Documents\Winpython64-3.9.10.0\WPy64-39100\python-3.9.10.amd64\lib\multiprocessing\process.py", line 315, in _bootstrap
    self.run()
  File "C:\Users\taichi\Documents\Winpython64-3.9.10.0\WPy64-39100\python-3.9.10.amd64\lib\multiprocessing\process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\taichi\Desktop\test.py", line 29, in p_test
    print(array.decode("utf-8"))
AttributeError: 'SynchronizedString' object has no attribute 'decode'

解决方法 #

错误核心原因：用Array('c')创建的共享内存对象是SynchronizedString类型，该对象本身没有decode方法。必须先通过切片array[:]获取字节数组，再调用decode方法，和第一个成功的Python案例逻辑保持一致。

需要修改两处代码：

1. p_test函数中的打印语句：

   print(array[:].decode("utf-8"))
   

2. TextWidget类的buttonClicked方法中的赋值语句：

   self.ids.lab.text = self.array[:].decode("utf-8")
   

修改后的完整代码如下：

from multiprocessing import Array,Process, Value, set_start_method, freeze_support, Lock
import math

kvLoadString=("""
<TextWidget>:
    BoxLayout:
        orientation: 'vertical'
        size: root.size

        Button:
            id: button1
            text: "start"
            font_size: 48
            on_press: root.buttonClicked()
        Label:
            id: lab
            text: 'result'
""")

def start_process(array,lock):
    process_test = Process(target=p_test, args=[array,lock], daemon=True)
    process_test.start()
    return process_test

def p_test(array,lock):
    with lock:
        buffer_data = str("z1111").encode("utf-8")
        array[0:len(buffer_data)] = buffer_data[:]
        print(array[:].decode("utf-8"))  # 修改此处

if __name__ == '__main__':
    freeze_support() 
    from kivy.app import App
    from kivy.uix.widget import Widget
    from kivy.lang import Builder

    from kivy.uix.checkbox import CheckBox
    from kivy.uix.spinner import Spinner
    from kivy.properties import StringProperty,NumericProperty,ObjectProperty, BooleanProperty
    from kivy.core.window import Window
    from kivy.uix.boxlayout import BoxLayout
    from kivy.uix.popup import Popup
    from kivy.uix.treeview import TreeView, TreeViewLabel, TreeViewNode
    from kivy.uix.label import Label
    from kivy.uix.textinput import TextInput
    from kivy.uix.progressbar import ProgressBar
    from kivy.uix.button import Button

    set_start_method('spawn')


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

        def build(self):
            return TextWidget()

    class TextWidget(Widget):
        def __init__(self, **kwargs):
            super(TextWidget, self).__init__(**kwargs)
            self.process_test = None
            self.proc = None

            # shared memory
            self.array = Array('c', 35)
            self.lock = Lock()

        def buttonClicked(self):
            if self.ids.button1.text == "start":
                self.proc = start_process(self.array,self.lock)
                self.ids.button1.text = "stop"
            else:
                if self.proc:
                    self.proc.kill()
                self.ids.button1.text = "start"
                self.ids.lab.text = self.array[:].decode("utf-8")  # 修改此处

    Builder.load_string(kvLoadString)
    TestApp().run()

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