一、 业务背景
在系统运维与监控场景中，通常需要同时对 20 个以上的 TCP 端口进行连通性检测。传统的 Shell 脚本采用串行检测方式，不仅耗时较长，且在异常处理与日志记录方面表现欠佳。为此，本文采用 Python 的 threading 模块实现单进程多线程并发检测，以大幅提升检测效率并规范日志输出。
二、 核心技术模块
1. 多线程并发控制
由于多线程共享内存资源，数据同步是并发编程的关键。本方案采用“列表作为任务队列”的轻量级设计：
•将待检测的端口列表作为共享资源。•每个线程通过 list.pop() 方法获取任务，只要列表非空就持续弹出并处理，列表为空则自动退出。•使用 threading.Thread 创建线程，并通过 join() 方法阻塞主线程，确保所有子线程执行完毕后再进行后续逻辑。
优化建议：虽然列表 pop() 在 CPython 中由于 GIL（全局解释器锁）的存在是线程安全的，但在高并发或复杂场景下，更推荐使用 Python 内置的 queue.Queue 模块，它提供了更完善的线程安全机制。
2. 外部命令调用
使用 subprocess 模块调用系统命令（如 nc）进行端口探测。
•该模块可以捕获并抑制命令的标准输出，避免控制台信息杂乱。•通过 subprocess.call() 的返回值判断执行状态：返回 0 表示命令执行成功（即端口连通），非 0 表示失败。
3. 日志记录
使用 logging 模块替代 print，实现结构化的日志管理。
•支持设置日志级别（如 DEBUG, INFO, ERROR）。•支持自定义输出格式（包含时间、文件名、行号等）。•支持将日志持久化到文件，便于事后排查。
三、 源码实现
以下代码在原版基础上进行了现代化重构，兼容 Python 3，并增强了异常处理与代码规范：
#!/usr/bin/env python3# -*- coding:utf-8 -*-"""TCP 端口并发检测脚本依赖环境：需安装 netcat (nc) 组件，如: yum install -y nmap-ncat"""import subprocessimport threadingimport osimport logging# ================= 配置区 =================HOST_SERVER = 'localhost'PORT_LIST = [ 9761, 9762, 9763, 9781, 9766, 9767, 9768, 9769, 9770, 9771, 8001, 8032, 8031, 8061, 8062, 8063, 8073, 8852, 8850, 8851, 8853, 8860]THREAD_COUNT = 10 # 并发线程数# 线程安全的结果收集列表port_alive = []port_unreacheable = []# 锁用于保护列表的写入操作result_lock = threading.Lock()# ================= 日志配置 =================def setup_logger(log_file): logging.basicConfig( level=logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s', datefmt='%Y-%m-%d %H:%M:%S', filename=log_file, filemode='a+', encoding='utf-8' # Python3 推荐指定编码 ) return logging.getLogger('PortChecker')# ================= 核心检测逻辑 =================def is_reachable(port_list, host, logger): """工作线程函数：循环从列表获取端口并检测""" while True: try: # 线程安全地获取端口 with result_lock: if not port_list: break port = port_list.pop() except IndexError: break try: # 使用 nc 命令检测端口，超时设为 4 秒 cmd = ["nc", "-v", "-w", "4", "-z", host, str(port)] ret_code = subprocess.call(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) if ret_code == 0: with result_lock: port_alive.append(port) logger.info(f'端口: {port} 正常') else: with result_lock: port_unreacheable.append(port) logger.error(f'端口: {port} 异常') except Exception as e: logger.error(f'端口: {port} 检测发生异常: {e}')# ================= 主程序入口 =================def main(): ab_path = os.path.abspath('./') res_log = os.path.join(ab_path, 'xone_status.log') logger = setup_logger(res_log) # 复制一份端口列表用于 pop 操作，避免修改原始配置 task_queue = PORT_LIST.copy() threads = [] for _ in range(THREAD_COUNT): thr = threading.Thread(target=is_reachable, args=(task_queue, HOST_SERVER, logger)) thr.start() threads.append(thr) # 等待所有线程完成 for thr in threads: thr.join() logger.info(f"检测完成。正常: {len(port_alive)}, 异常: {len(port_unreacheable)}") # 若存在异常端口，返回非零退出码，便于外部脚本捕获 if port_unreacheable: exit(4)if __name__ == '__main__': main()
四、 核心优化点说明
1.线程安全：引入了 threading.Lock()。虽然 list.pop() 在 CPython 中是原子的，但向 port_alive 追加数据以及判断列表是否为空的操作组合在一起时并非绝对安全，加锁可避免潜在的竞态条件。2.命令调用安全：将 subprocess.call 的字符串参数改为列表形式（如 ["nc", "-v", ...]），并设置 shell=False（默认值），这能有效防止命令注入风险，且避免了依赖 shell=True 带来的额外开销。3.输出抑制：使用 stdout=subprocess.DEVNULL 替代 shell=True 下的重定向，更加 Pythonic 且跨平台兼容性更好。4.Python 3 兼容：将 format() 字符串替换为 f-string，增加了 encoding='utf-8' 参数，确保日志写入不乱码。
五、 运行效果与日志分析
执行脚本后，可通过以下命令实时查看检测日志：
预期日志输出示例：
2020-07-14 09:49:56 xone_check.py[line:68] INFO 端口:9781 正常2020-07-14 09:49:56 xone_check.py[line:68] INFO 端口:9770 正常2020-07-14 09:49:56 xone_check.py[line:73] ERROR 端口:8888 异常...
性能收益： 得益于多线程并发，20+ 个端口的检测耗时从串行模式的数十秒缩短至 1~2 秒内完成，显著提升了运维巡检效率。
