Kivy与multiprocessing结合时进程阻塞无响应问题求助

Hey, I've run into this exact issue before with Kivy and multiprocessing on Windows—let me break down what's happening and how to fix it.

The core problem is a conflict between Kivy's app initialization and how multiprocessing works on Windows (using the spawn start method by default). When you launch a subprocess from Kivy's UI callback:

• Windows creates a new process by re-importing your entire script.
• If your Kivy app startup code isn't properly guarded, the subprocess will try to initialize a new Kivy app instance, which gets stuck and never runs your pprosess function.
• On top of that, calling q.get() directly in the UI thread blocks the entire interface—even if the subprocess worked, your app would freeze until the message arrives.

Here's the adjusted code that fixes both issues, with comments explaining each change:

from kivy.app import App
from kivy.lang import Builder
from kivy.uix.boxlayout import BoxLayout
from kivy.clock import Clock
from multiprocessing import Process, Queue, Event

kv = '''
<InterfaceView>:
    Button:
        text: 'test'
        on_press: root.do_process()
'''

# Keep your subprocess function at the top so it's accessible to child processes
def pprocess(message, q, stop):
    q.put(message)
    stop.wait()

Builder.load_string(kv)

class InterfaceView(BoxLayout):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.message_queue = None
        self.stop_event = None

    def do_process(self):
        self.message_queue = Queue()
        self.stop_event = Event()
        
        # Launch the subprocess
        p = Process(target=pprocess, args=('any message', self.message_queue, self.stop_event))
        p.daemon = True
        p.start()
        print('Waiting for message...')
        
        # Use Kivy's Clock to poll the queue non-blockingly
        Clock.schedule_interval(self.check_for_message, 0.1)

    def check_for_message(self, dt):
        # Only check if the queue exists and has content
        if self.message_queue and not self.message_queue.empty():
            message = self.message_queue.get()
            print('message: ', message)
            
            # Signal the subprocess to exit
            self.stop_event.set()
            # Stop polling the queue
            Clock.unschedule(self.check_for_message)
            print('Process closed')

class SimpleApp(App):
    def build(self):
        return InterfaceView()

# CRITICAL: Guard Kivy's startup code so subprocesses don't run it
if __name__ == '__main__':
    SimpleApp().run()

Key Fixes: #

1. if __name__ == '__main__' Guard: When Windows spawns a subprocess, it re-imports your script. This guard ensures the Kivy app only launches in the main process—subprocesses skip this part and just load the pprocess function.
2. Non-Blocking Queue Polling: Using Clock.schedule_interval lets us check the queue without freezing the UI. This is a best practice for all Kivy UI interactions that might wait for external data.

Your non-Kivy code worked because the subprocess only loaded the pprocess function and never tried to run the main() function (thanks to the same __name__ guard).

• On Linux/macOS, multiprocessing uses fork by default, so the guard isn't strictly necessary—but it's still good practice for cross-platform compatibility.
• Never run blocking calls (like q.get(), time.sleep()) directly in Kivy's UI thread—always use Clock or separate threads/processes with non-blocking checks.

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