Python多线程本地IP扫描器KeyboardInterrupt捕获无效如何解决

核心问题原因 #

• 原代码主进程阻塞在q.join()调用，该状态下默认无法响应KeyboardInterrupt异常
• 没有设置线程终止通知机制，子线程会持续消费队列中的任务直到全部完成
• ping调用没有设置超时，可能出现进程阻塞导致脚本无法及时退出
• 原代码重复执行两次ping命令，也会拉长扫描耗时导致终止延迟

具体修改方案 #

关键优化点 #

1. 新增threading.Event类型的终止事件，用来通知所有子线程停止工作
2. 将阻塞式q.join()改为带超时的循环等待，确保可以捕获Ctrl+C信号
3. 捕获到终止信号后立即清空任务队列，触发终止事件，快速结束所有线程运行
4. 合并两次ping调用为一次，增加超时参数，避免进程卡死
5. 子线程的循环逻辑中增加终止事件检查，收到终止信号后立即退出

完整可运行代码 #

import threading
import queue
import subprocess
import platform
import socket
import datetime
import time
import sys
from colorama import Back, Fore, Style

# 初始化队列、锁、终止事件
q = queue.Queue()
lock = threading.Lock()
exit_event = threading.Event()

def ping(ip):
    if exit_event.is_set():
        return
    if platform.system().lower() == 'windows':
        param = '-n'
        encoding = 'gbk'
    else:
        param = '-c'
        encoding = 'utf-8'
    command = ['ping', param, '1', '-w', '1000', ip]
    try:
        result = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        output = result.communicate(timeout=2)[0].decode(encoding, errors='ignore').lower()
    except subprocess.TimeoutExpired:
        result.kill()
        return
    if 'destination host unreachable' in output or 'request timed out' in output or '100% 丢失' in output or '100% packet loss' in output:
        return
    else:
        with lock:
            try:
                host = socket.gethostbyaddr(ip)[0]
            except:
                host = 'No hostname'
            now = datetime.datetime.now()
            print(Back.LIGHTGREEN_EX + Fore.BLACK + now.strftime('[%Y-%m-%d %H:%M:%S]') + Style.RESET_ALL + Fore.LIGHTGREEN_EX + f' {host} | {ip} | Active')

def threader():
    while not exit_event.is_set():
        try:
            worker = q.get(timeout=0.5)
            ping(worker)
            q.task_done()
        except queue.Empty:
            continue

if __name__ == '__main__':
    # 启动线程
    for x in range(60):
        t = threading.Thread(target=threader)
        t.daemon = True
        t.start()
    
    # 获取本地IP前缀
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.connect(('8.8.8.8', 53))
    ip = s.getsockname()[0]
    s.close()
    prefix = ip.split('.')
    prefix.pop()
    base_ip = '.'.join(prefix) + '.'
    
    # 填充扫描任务
    for worker in range(256):
        q.put(base_ip + str(worker))
    
    # 等待任务完成，支持Ctrl+C终止
    try:
        while not q.empty() and not exit_event.is_set():
            time.sleep(0.5)
    except KeyboardInterrupt:
        print("\n正在终止扫描...")
        exit_event.set()
        # 清空剩余任务
        while not q.empty():
            try:
                q.get(block=False)
                q.task_done()
            except queue.Empty:
                break
        print("扫描已停止")
    finally:
        print(Style.RESET_ALL, end='')
        sys.exit(0)

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