Python多线程致Windows键盘输入延迟，求更优解决方案

问题场景 #

在Windows上运行Python脚本，创建5个线程执行轻量任务（API调用、简单计算），但脚本运行时键盘输入有明显延迟。移除一个线程后（4个线程）一切正常，但业务需要必须用满5个线程。

推测原因 #

系统键盘事件监听进程和Python脚本进程共用同一CPU核心，脚本的多线程持续占用核心资源，导致键盘监听线程的可用CPU时间被挤占，出现输入延迟。

当前临时解决方法 #

通过psutil设置Python进程的CPU亲和性，绑定到特定核心后问题解决：

p = psutil.Process()
p.cpu_affinity([1])

但不确定这是不是最优方案，想找更好的解决思路。

核心代码逻辑（简化版） #

脚本包含4个核心线程（加上热键线程共5个）：

• reader线程：持续监听用户输入，预处理后将消息加入对应列表list_A或list_B
• response_generator线程：读取消息列表，调用API生成响应后加入list_C
• response_execution线程：读取list_C，执行响应（涉及API调用）
• stop_robot_thread：通过全局热键<ctrl>+<alt>触发应急停止

代码如下：

def reader():
    global list_A
    global list_B
    # 持续检查输入消息并评估
    # 预处理消息（格式化等）
    # 消息合法则根据类型加入list_A或list_B


def response_generator():
    global list_A
    global list_B
    global list_C

    while True:
        if list_A:
            # 处理list_A消息，调用API生成响应
            response = generate_response_A()
        if list_B:
            # 处理list_B消息，调用API生成响应
            response = generate_response_B()
        # 将响应加入list_C
        list_C.append(response)


def response_execution():
    global list_C

    while True:
        if list_C:
            # 执行响应，涉及API调用
            pass


def on_activate_h():
    # 立即取消执行
    robot.stop_execution()


input_thread = Thread(target=reader)
response_generator_thread = Thread(target=response_generator)
response_execution_thread = Thread(target=response_execution)
stop_robot_thread = keyboard.GlobalHotKeys({
    '<ctrl>+<alt>': on_activate_h})

input_thread.start()
response_generator_thread.start()
response_execution_thread.start()
stop_robot_thread.start()

input_thread.join()
response_generator_thread.join()
response_execution_thread.join()
stop_robot_thread.join()

优化方案建议 #

1. 替换轮询为阻塞队列，减少空循环CPU消耗 #

当前response_generator和response_execution用while True空轮询列表，会持续占用CPU。改用Python标准库的queue.Queue（线程安全），利用其阻塞特性（get()方法默认会等待队列有数据），彻底消除空轮询的资源浪费：

from queue import Queue

# 初始化队列替代全局列表
queue_A = Queue()
queue_B = Queue()
queue_C = Queue()

def reader():
    # 预处理后放入对应队列
    queue_A.put(processed_msg_A)
    # 或 queue_B.put(processed_msg_B)

def response_generator():
    while True:
        # 阻塞等待queue_A的消息
        msg = queue_A.get()
        response = generate_response_A()
        queue_C.put(response)
        
        # 同理处理queue_B
        msg = queue_B.get()
        response = generate_response_B()
        queue_C.put(response)

def response_execution():
    while True:
        response = queue_C.get()
        # 执行响应逻辑

2. 提升输入/热键线程的优先级 #

在Windows上，通过ctypes调用系统API，把reader和热键线程的优先级调高，确保键盘相关的线程能优先获得CPU时间，避免被其他线程挤占：

import ctypes
from threading import Thread

def set_thread_high_priority(thread):
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    # 最高线程优先级（Windows定义的常量）
    THREAD_PRIORITY_HIGHEST = 2
    # 设置线程优先级
    kernel32.SetThreadPriority(thread.ident, THREAD_PRIORITY_HIGHEST)

# 启动线程后调用
input_thread.start()
set_thread_high_priority(input_thread)

stop_robot_thread.start()
# 注意：keyboard库的线程需获取其内部线程对象设置优先级
set_thread_high_priority(stop_robot_thread.thread)

3. 精细化CPU亲和性分配 #

不用把整个进程绑定到单个核心，而是给键盘相关线程单独绑定到空闲核心，其他线程绑定到剩余核心，更合理利用CPU资源：

import ctypes
from threading import Thread

def set_thread_affinity(thread, cores):
    kernel32 = ctypes.WinDLL('kernel32')
    # 生成核心掩码
    core_mask = 0
    for core in cores:
        core_mask |= 1 << core
    # 设置线程亲和性
    kernel32.SetThreadAffinityMask(thread.ident, core_mask)

# 启动输入线程后绑定到核心0
input_thread.start()
set_thread_affinity(input_thread, [0])
# 其他线程绑定到核心1-3
response_generator_thread.start()
set_thread_affinity(response_generator_thread, [1,2,3])

4. 移除全局变量，用线程安全容器封装状态 #

全局变量的无锁访问可能引发隐性的线程竞争和额外开销，改用类封装所有状态和队列，代码更清晰且减少不必要的CPU消耗：

from queue import Queue

class RobotService:
    def __init__(self):
        self.queue_A = Queue()
        self.queue_B = Queue()
        self.queue_C = Queue()
        self.running = True

    def reader(self):
        while self.running:
            # 处理输入，放入对应队列
            pass

    def response_generator(self):
        while self.running:
            msg = self.queue_A.get()
            resp = generate_response_A(msg)
            self.queue_C.put(resp)
            # 处理queue_B同理

    def response_execution(self):
        while self.running:
            resp = self.queue_C.get()
            # 执行响应

    def stop(self):
        self.running = False
        robot.stop_execution()

# 初始化并启动线程
service = RobotService()
input_thread = Thread(target=service.reader)
# ... 其他线程初始化和启动

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