多线程结合input()实现优雅循环退出的问题排查与优化

问题根源 #

当发送线程正常结束后，输入监听线程仍阻塞在input()调用中，此时关闭程序会触发输入流解码错误；同时未处理线程间的优雅同步，导致需要多次终止操作才能关闭程序。

改进方案：使用线程事件+跨平台非阻塞输入 #

1. 替换全局变量为线程事件 #

用threading.Event替代全局stop_flag，提供更安全的线程同步机制，避免全局变量的副作用。

2. 实现跨平台非阻塞输入 #

封装跨平台函数，避免监听线程长期阻塞在input()：

import sys
import select
import msvcrt

def has_available_input(timeout=0.1):
    """跨平台检查是否有输入等待读取"""
    if sys.platform == 'win32':
        return msvcrt.kbhit()
    else:
        ready, _, _ = select.select([sys.stdin], [], [], timeout)
        return ready != []

def read_user_input():
    """跨平台读取用户输入"""
    if sys.platform == 'win32':
        # Windows下读取一行（处理回车）
        input_chars = []
        while True:
            char = msvcrt.getwche()
            if char == '\r':
                print()
                break
            input_chars.append(char)
        return ''.join(input_chars)
    else:
        return sys.stdin.readline().strip()

3. 重构发送与监听函数 #

import threading
import time

def send_messages(count_msg: int, delay_msg: int, stop_event: threading.Event, done_event: threading.Event):
    try:
        for msg_idx in range(count_msg):
            if stop_event.is_set():
                print('---Sending completed by the user [ОК]---')
                break
            # 替换为你的实际发送逻辑
            print(f"Sent message {msg_idx + 1}")
            time.sleep(delay_msg)
    finally:
        # 标记发送任务完成，通知监听线程退出
        done_event.set()
        # 重置停止事件，方便后续复用函数
        if stop_event.is_set():
            stop_event.clear()

def monitor_input(stop_event: threading.Event, done_event: threading.Event):
    # 循环监听：直到发送任务完成或收到stop指令
    while not done_event.is_set():
        if has_available_input():
            user_input = read_user_input()
            if user_input.strip().lower() == 'stop':
                stop_event.set()
                break
        # 短暂休眠，避免占用过多CPU资源
        time.sleep(0.1)

4. 主线程控制逻辑 #

if __name__ == "__main__":
    # 初始化线程事件
    stop_event = threading.Event()
    done_event = threading.Event()

    # 创建并启动线程
    send_thread = threading.Thread(
        target=send_messages,
        args=(5, 1, stop_event, done_event)  # 替换为你的实际参数
    )
    monitor_thread = threading.Thread(
        target=monitor_input,
        args=(stop_event, done_event)
    )

    try:
        send_thread.start()
        monitor_thread.start()

        # 等待线程完成
        send_thread.join()
        monitor_thread.join()
    except KeyboardInterrupt:
        # 处理Ctrl+C终止，优雅停止所有线程
        stop_event.set()
        done_event.set()
        print("\nProgram terminated by user")

    print("Program exited normally")

关键改进点 #

• 线程同步：用done_event通知监听线程发送任务已完成，避免监听线程长期阻塞
• 非阻塞输入：跨平台实现输入检查，既保留监听功能，又能在任务结束后自动退出
• 优雅终止：捕获KeyboardInterrupt，确保所有线程能收到停止信号并退出
• 无全局变量：使用线程事件传递状态，代码更安全、可维护

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