Pyautogui结合Xlib与多进程触发RuntimeError问题求助

这个RuntimeError是X11协议通信混乱导致的：

• Linux下PyAutoGUI依赖Xlib与X服务器交互，每个进程的X11连接有独立的请求序列号。
• 当子进程继承主进程的X11连接、或进程被粗暴终止时，X服务器返回的响应序列号和当前进程预期的不匹配，触发协议错误。

• 多进程直接调用PyAutoGUI：子进程继承主进程的X11连接，多个进程同时操作鼠标/查询位置，打乱X服务器的请求响应序列。
• 错误的进程终止方式：os.kill(t2, __signal=(exit()))、t1.kill()完全错误——exit()不是合法信号，粗暴终止进程会导致Xlib请求队列未清理，残留序列号引发不匹配。
• Event对象使用错误：while not event:直接判断Event对象永远为True，正确写法是event.is_set()；当前逻辑完全没实现“一个进程完成后终止另一个”的需求。
• 主进程存活判断错误：if t1.is_alive是判断方法对象是否存在（永远为True），必须调用t1.is_alive()方法。
• 低效资源加载：每个循环都重新读取模板图片，浪费IO；两个子进程重复抓取屏幕，无意义消耗CPU。
• 坐标计算错误：模板中心坐标计算时宽度和高度顺序搞反，导致鼠标定位偏移。

1. 主进程统一处理PyAutoGUI操作：子进程只做模板匹配，把匹配结果（坐标）传递给主进程，由主进程执行鼠标移动，避免多进程操作X11连接。
2. 正确终止进程：用process.terminate()或发送合法信号（如signal.SIGTERM），避免粗暴终止。
3. 正确使用Event控制流程：用event.is_set()判断是否终止循环，实现进程间同步。
4. 预加载模板：进程启动前读取模板图片，避免循环重复加载。
5. 统一屏幕抓取：主进程负责抓取屏幕并传递给子进程，避免重复抓取。

修复后代码示例 #

import cv2 as cv
import numpy as np
from PIL import ImageGrab
import pyautogui
import multiprocessing
import time
import signal

# 预加载模板图片
TEMPLATE1_PATH = '/run/media/marcos/8d82f6aa-2fc2-4eb7-ba62-e714d909f394/Old files/PycharmProjects/REDLIGHT_automation/Video_player_end.png'
TEMPLATE2_PATH = "/home/marcos/Pictures/reject_window.png"

template1 = cv.cvtColor(cv.imread(TEMPLATE1_PATH), cv.COLOR_BGR2GRAY)
template2 = cv.cvtColor(cv.imread(TEMPLATE2_PATH), cv.COLOR_BGR2GRAY)
w1, h1 = template1.shape[::-1]
w2, h2 = template2.shape[::-1]

# 匹配区域
REGION = (796, 574, 1756, 973)

def match_template_1(event, result_queue, screen_queue):
    while not event.is_set():
        screen_gray = screen_queue.get()
        if screen_gray is None:
            break
        
        # 执行模板匹配
        result = cv.matchTemplate(screen_gray, template1, cv.TM_CCORR_NORMED)
        _, max_val, _, max_loc = cv.minMaxLoc(result)
        
        if max_val >= 0.95:
            # 计算正确的模板中心坐标
            center_x = max_loc[0] + w1 / 2
            center_y = max_loc[1] + h1 / 2
            result_queue.put(('template1', center_x, center_y))
            event.set()
            break

def match_template_2(event, result_queue, screen_queue):
    while not event.is_set():
        screen_gray = screen_queue.get()
        if screen_gray is None:
            break
        
        result = cv.matchTemplate(screen_gray, template2, cv.TM_CCORR_NORMED)
        _, max_val, _, max_loc = cv.minMaxLoc(result)
        
        if max_val >= 0.98:
            center_x = max_loc[0] + w2 / 2
            center_y = max_loc[1] + h2 / 2
            result_queue.put(('template2', center_x, center_y))
            event.set()
            break

def test1():
    print('sleeping for 5 sec....')
    pyautogui.moveTo(800,800)
    time.sleep(5)
    print('done sleeping 1')

if __name__ == "__main__":
    event = multiprocessing.Event()
    result_queue = multiprocessing.Queue()
    screen_queue = multiprocessing.Queue()

    # 启动匹配进程
    p1 = multiprocessing.Process(target=match_template_1, args=(event, result_queue, screen_queue))
    p2 = multiprocessing.Process(target=match_template_2, args=(event, result_queue, screen_queue))
    p1.start()
    p2.start()

    try:
        # 主进程循环抓取屏幕并传递给子进程
        while not event.is_set():
            screen = np.array(ImageGrab.grab(bbox=(0, 0, 2560, 1440)))
            screen_gray = cv.cvtColor(screen, cv.COLOR_RGB2GRAY)
            # 裁剪到指定匹配区域
            x, y, w, h = REGION
            screen_gray_cropped = screen_gray[y:y+h, x:x+w]
            
            # 发送给两个子进程
            screen_queue.put(screen_gray_cropped.copy())
            screen_queue.put(screen_gray_cropped.copy())
            
            time.sleep(0.1)  # 降低循环频率，减少CPU占用
    except KeyboardInterrupt:
        # 捕获中断信号，清理进程
        event.set()
        screen_queue.put(None)
        screen_queue.put(None)

    # 获取匹配结果
    result = result_queue.get()
    # 终止并回收子进程
    p1.terminate()
    p2.terminate()
    p1.join()
    p2.join()

    # 主进程执行鼠标移动
    if result:
        _, cx, cy = result
        pyautogui.moveTo(cx, cy)
        print(f"匹配成功，鼠标移动到({cx:.2f}, {cy:.2f})")

    # 执行后续逻辑
    if result[0] == 'template2':
        print('线程TWO匹配成功，线程ONE已终止')
        test1()
    else:
        print('线程ONE匹配成功，线程TWO已终止')

内容的提问来源于stack exchange，提问作者Marcos Umana Santos