Python使用threading+Telnet多线程连接多台交换机查询CPU占用异常问题

核心错误原因 #

• 全局仅创建了1个telnetlib.Telnet实例，且初始仅连接了第一台交换机192.168.1.42，两个线程同时操作同一个连接对象，命令发送和IO读写互相串扰，第二台交换机根本没有建立独立连接
• switch2函数中用到的password变量未提前定义，执行会报错
• Telnet连接不是线程安全的，多线程并发操作同一个连接必然会出现数据异常

修正后代码 #

import time
import telnetlib
import threading

# 所有交换机配置统一维护，后续新增设备直接加配置即可
device_configs = [
    {
        "host": "192.168.1.42",
        "username": "root",
        "password": "替换为第一台交换机实际密码"
    },
    {
        "host": "192.168.86.247",
        "username": "root",
        "password": "替换为第二台交换机实际密码"
    }
]

def query_cpu(device):
    host = device["host"]
    try:
        # 每个线程独立创建专属Telnet连接，互不干扰
        tn = telnetlib.Telnet(host, timeout=10)
        # 输入用户名
        tn.write(device["username"].encode("ascii") + b"\n")
        time.sleep(0.5)
        # 输入密码
        tn.write(device["password"].encode("ascii") + b"\n")
        time.sleep(0.5)
        print(f"成功连接交换机：{host}")
        # 发送CPU查询命令
        tn.write(b"sh cpu-usage\n")
        time.sleep(2)
        # 读取并解码输出
        output = tn.read_very_eager().decode("ascii", errors="ignore")
        print(f"\n===== {host} CPU查询结果 =====")
        print(output)
        print(f"{host} 采集完成\n")
        # 关闭连接
        tn.close()
    except Exception as e:
        print(f"交换机{host}采集失败：{str(e)}")

if __name__ == "__main__":
    threads = []
    # 批量创建线程
    for dev in device_configs:
        t = threading.Thread(target=query_cpu, args=(dev,))
        threads.append(t)
        t.start()
    # 等待所有线程执行结束
    for t in threads:
        t.join()

优化说明 #

• 每个交换机对应独立的Telnet连接，线程间无共享资源，完全避免读写冲突
• 通用采集逻辑封装为函数，新增设备无需重复写代码，仅需修改配置列表即可
• 增加超时设置和异常捕获，避免单个设备故障导致整个程序卡住
• 输出结果默认做解码处理，可读性更强

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