Keyboard库事件阻塞执行：Socket传输脚本超时未触发处理函数

你的代码核心问题是keyboard.read_event(suppress=True)是阻塞式调用——程序会一直卡在这行代码，直到有键盘事件发生才会继续执行后续逻辑。这就导致超时判断的代码只有在有键盘按键按下时才会被检查，完全无法主动触发超时逻辑。只要没有新的键盘输入，程序就永远不会走到超时处理的分支里。

下面提供两种可行的修改思路，解决阻塞导致的超时失效问题：

方案1：按键回调+主线程超时检查 #

通过回调函数记录按键输入，主线程循环检查超时状态，避免阻塞：

import socket
import keyboard
import time

# TCP/IP server details
server_ip = "10.10.0.0"
server_port = 5200

# Timeout for barcode input in seconds
timeout_duration = 0.5

barcode = ""
last_key_time = time.time()

def send_data(data):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect((server_ip, server_port))
        s.sendall(data.encode())

def process_barcode(barcode):
    if ';' in barcode:
        forwarded_data = f"{barcode};1"
    else:
        forwarded_data = f"test;{barcode};1"
    
    send_data(forwarded_data)

def on_key_press(event):
    global barcode, last_key_time
    if event.event_type == keyboard.KEY_DOWN:
        # 忽略扫码枪自动发送的回车（按需调整）
        if event.name == 'enter':
            return
        barcode += event.name
        last_key_time = time.time()
        print(f"当前输入: {barcode}")

def main():
    global barcode, last_key_time
    # 注册按键监听回调
    keyboard.on_press(on_key_press, suppress=True)
    
    print("等待条码输入...")
    
    while True:
        current_time = time.time()
        # 检查超时且已有输入
        if (current_time - last_key_time > timeout_duration) and barcode:
            print("触发超时处理")
            process_barcode(barcode)
            # 重置输入状态
            barcode = ""
            last_key_time = current_time
        # 短暂休眠降低CPU占用
        time.sleep(0.01)

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("程序终止")

方案2：带超时参数的事件读取 #

如果你的keyboard库版本支持，直接给read_event添加超时参数，超时后抛出异常并处理：

import socket
import keyboard
import time

# TCP/IP server details
server_ip = "10.10.0.0"
server_port = 5200

# Timeout for barcode input in seconds
timeout_duration = 0.5

def send_data(data):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect((server_ip, server_port))
        s.sendall(data.encode())

def process_barcode(barcode):
    if ';' in barcode:
        forwarded_data = f"{barcode};1"
    else:
        forwarded_data = f"test;{barcode};1"
    
    send_data(forwarded_data)

def main():
    barcode = ""
    last_key_time = time.time()
    
    print("等待条码输入...")
    
    while True:
        try:
            # 设置超时时间，超时会抛出TimeoutError
            event = keyboard.read_event(suppress=True, timeout=timeout_duration)
            if event.event_type == keyboard.KEY_DOWN:
                if event.name == 'enter':
                    continue
                barcode += event.name
                print(barcode)
                last_key_time = time.time()
        except keyboard.TimeoutError:
            # 超时触发，检查是否有输入需要处理
            if barcode:
                print("触发超时处理")
                process_barcode(barcode)
                barcode = ""

if __name__ == "__main__":
    main()

• 扫码枪通常会在扫描完成后自动发送回车按键，可以直接监听回车事件触发条码处理，比超时判断更准确。
• 可以添加退格键处理逻辑，避免输入错误的条码数据。

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