如何在Python多进程Process中检测按键并安全终止程序运行

问题根因 #

1. 生产数据的循环没有加任何延时，程序会以最高频率往队列塞随机数据，按键触发时队列已经积压了大量未处理数据，消费进程要处理完所有积压数据才能收到终止用的None，视觉上就像终止逻辑失效，同时大量的输出也会干扰调试。
2. 全局键盘监听库需要系统权限（Windows需要管理员权限、macOS需要开启辅助功能权限），放在子进程中运行时很容易出现权限不足、响应延迟的问题，本身设计就不适合在子进程中执行。

优化方案 #

把键盘监听逻辑移到主进程，配合multiprocessing.Event做全局终止通知，避免子进程权限问题，同时给生产循环加小间隔控制数据生产速度，解决队列积压问题。
修改后可运行代码如下：

import multiprocessing
import keyboard
import numpy as np
import time

def getData(queue, stop_event):
    while not stop_event.is_set():
        # 每次生产前先检查终止信号
        if stop_event.is_set():
            break
        data = np.random.random([8, 250])
        queue.put(data)
        # 控制生产速度，可根据实际需求调整间隔
        time.sleep(0.01)
    # 收到终止信号后放None通知消费进程退出
    queue.put(None)
    print("STOP in getData")

def processData(queue):
    while True:
        data = queue.get()
        if data is None:
            print("STOP in processData")
            break
        print("Processing Data")
        print(str(data[0,0]))

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    stop_event = multiprocessing.Event()
    
    processes = [
        multiprocessing.Process(target=getData, args=(queue, stop_event)),
        multiprocessing.Process(target=processData, args=(queue,))
    ]

    for p in processes:
        p.start()
    
    # 主进程监听按键
    keyboard.wait("a")
    # 按下a后发送终止信号给生产进程
    stop_event.set()

    for p in processes:
        p.join()

补充说明 #

如果需要使用pynput方案，逻辑完全一致，只需要把主进程的keyboard.wait("a")替换成pynput的按键监听逻辑即可，监听逻辑都放在主进程就不会出现子进程权限异常的问题。

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