如何按'q'键终止多线程自动Ping路由器脚本？

问题分析 #

你的脚本存在几个关键问题导致无法正常终止：

• qu函数里的os._exit仅引用了函数，未实际调用，正确写法应为os._exit(0)
• 主线程main1的循环无终止条件，即便监听线程触发退出，主循环仍会持续执行
• keyboard.is_pressed的无限循环会占用大量CPU资源，实现方式不够优雅
• 未将线程设置为守护线程，监听线程退出后主线程可能继续运行

修复后的代码 #

我们使用threading.Event实现优雅的终止控制，同时优化ping操作的中断逻辑，避免长时间阻塞：

import os
import time
import threading
import keyboard
import pandas as pd

# 全局终止事件，用于通知所有线程停止执行
stop_event = threading.Event()

def start():
    for file_name in file_list:
        # 每次循环前检查终止信号
        if stop_event.is_set():
            break
        unrechable = []
        rechable = []
        print(f"Processing: {file_name}")
        open_file(file_name, rechable, unrechable)

        if len(unrechable) > 0:
            print(f"These IP from {file_name} are Unreachable:")
            for i in unrechable:
                print(i)
            print("")
        else:
            print(f"All IP's are Reachable from {file_name}")

def open_file(file_name, rechable, unrechable):
    df = pd.read_excel(f"D:/Network/{file_name}.xlsx")
    col_IP = df.loc[:, "IP"].tolist()
    col_name = df.loc[:, "Location"].tolist()
    check(col_IP, col_name, rechable, unrechable)

def check(col_IP, col_name, rechable, unrechable):
    for idx in range(len(col_IP)):
        # 每次ping前检查终止信号
        if stop_event.is_set():
            break
        ip = col_IP[idx]
        location = col_name[idx]
        # 优化ping命令，限制发包数和超时时间，避免阻塞
        response = os.popen(f"ping {ip} -n 1 -w 2000").read()
        if "Request timed out." in response or "unreachable" in response:
            print(response)
            unrechable.append(f"{ip} at {location}")
        else:
            print(response)
            rechable.append(f"{ip} at {location}")

def main1():
    while not stop_event.is_set():
        start()
        if stop_event.is_set():
            break
        print("Going to Sleep for 2Hrs")
        # 用wait替代sleep，可被终止事件中断，无需等满2小时
        stop_event.wait(7200)  # 7200秒=2小时

def qu():
    # 用keyboard.wait替代轮询，降低CPU占用
    keyboard.wait('q')
    print("exit")
    stop_event.set()
    # 强制退出（防止主线程未及时响应事件）
    os._exit(0)

file_list = ["ISP", "NVR"]

if __name__ == '__main__':
    # 设置主线程为守护线程，主程序退出时自动终止
    p2 = threading.Thread(target=main1, daemon=True)
    p3 = threading.Thread(target=qu)

    p2.start()
    p3.start()
    # 等待监听线程结束，避免主程序提前退出
    p3.join()

关键优化点 #

• 用threading.Event作为全局终止信号，所有线程可通过检查该信号停止执行
• 将time.sleep(7200)替换为stop_event.wait(7200)，按下'q'时可立即唤醒主线程，无需等待满2小时
• 在start和check函数的循环中加入终止检查，确保正在执行的ping任务能及时停止
• 使用keyboard.wait('q')替代无限循环轮询，减少CPU资源消耗
• 将主线程设为守护线程，确保监听线程退出后整个程序能正常终止
• 优化ping命令参数，添加-n 1（只发一个包）和-w 2000（超时2秒），避免单个ping操作耗时过长

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