如何在Python调度任务中复用Netmiko连接至Cisco交换机

我尝试用Python的Netmiko SSH库每隔30秒从Cisco交换机获取信息，但短时间内新建过多连接会被交换机拒绝。我想用单例模式在多次调度任务中复用现有连接，熟悉Java的单例实现，但不清楚Python里的正确做法。以下是我的相关代码，恳请指导如何正确实现跨任务保持连接：

调度任务代码 #

def job():
    print("Job started")
    switch.disableAllUnsedPorts()
    print("Job done")

schedule.every(int(config['sheduler.general']['shedulerJobInSeconds'])).seconds.do(job)

def runScheduler():
    while 1:
        schedule.run_pending()
        time.sleep(1)

switchcontroller.py代码 #

def disableAllPorts():
    savePortStatusToDatabase(getPortStatusFromSwitch())

    for port in ports:
        disableSinglePort(port)
    ...

def disableSinglePort(port):
   config_commands = [
        "interface " + port["portname"], 
        "shutdown"
    ]
    SwitchConnection.getConnection().send_config_set(config_commands)
    ...

switchconnection.py代码 #

class SwitchConnection:
    __instance = None
    __connection = None

    @staticmethod
    def getInstance():
        if SwitchConnection.__instance == None:
            SwitchConnection()
        return SwitchConnection.__instance

    @staticmethod
    def getConnection():
        return SwitchConnection.__connection

    def __init__(self):
        """ Virtually private constructor. """
        if SwitchConnection.__instance != None:
            raise Exception("This class is a singleton!")
        else:
            SwitchConnection.__instance = self

    def connect(self, **kwargs):
        if self.__connection == None:
            self.__connection = ConnectHandler(**kwargs)
        return self.__connection

你的现有单例实现存在几个问题：

• getConnection是静态方法，但连接实例绑定到类实例，逻辑不匹配
• 缺少连接有效性检测，SSH连接长时间闲置可能被交换机主动断开
• 初始化逻辑冗余，Python有更简洁的单例实现方式

修正后的SwitchConnection单例实现 #

from netmiko import ConnectHandler
from netmiko.exceptions import NetMikoAuthenticationException, NetMikoTimeoutException

class SwitchConnection:
    _instance = None
    _connection = None

    def __new__(cls, *args, **kwargs):
        # Python用__new__实现单例更简洁直接
        if not cls._instance:
            cls._instance = super().__new__(cls)
        return cls._instance

    def get_connection(self, **device_params):
        # 检查连接是否存在且可用，不可用时自动重建
        if self._connection is None or not self._is_connection_alive():
            try:
                self._connection = ConnectHandler(**device_params)
                # 发送简单命令验证连接有效性
                self._connection.send_command("show version | include Version")
            except (NetMikoAuthenticationException, NetMikoTimeoutException) as e:
                print(f"连接交换机失败: {str(e)}")
                self._connection = None
        return self._connection

    def _is_connection_alive(self):
        # 内部方法：检测连接存活状态
        if self._connection is None:
            return False
        try:
            # 通过发送空字符快速检测连接
            self._connection.send_command("", delay_factor=0.1)
            return True
        except Exception:
            return False

    def disconnect(self):
        # 主动关闭连接，避免残留无效连接
        if self._connection:
            self._connection.disconnect()
            self._connection = None

使用方式调整 #

1. 在调度任务启动前初始化连接（或第一次调用时自动触发）：

# 初始化单例连接实例
switch_conn = SwitchConnection()
# 从配置中读取交换机参数
device_params = {
    "device_type": "cisco_ios",
    "host": config['switch']['host'],
    "username": config['switch']['username'],
    "password": config['switch']['password'],
    "secret": config['switch']['enable_password']  # 若需要进入特权模式
}
# 提前建立有效连接
switch_conn.get_connection(**device_params)

2. 修改switchcontroller.py中的调用逻辑：

def disableSinglePort(port):
   config_commands = [
        "interface " + port["portname"], 
        "shutdown"
    ]
    # 获取单例模式下的有效连接
    conn = SwitchConnection().get_connection()
    if conn:
        conn.send_config_set(config_commands)
    else:
        print("无法获取有效连接，跳过端口关闭操作")
    ...

额外注意事项 #

• 即便复用连接，若任务间隔超过交换机SSH超时时间，连接仍可能被断开，get_connection里的存活检测会自动重建连接
• 程序退出时建议调用disconnect()主动关闭连接，避免留下无效连接资源
• 若需管理多台交换机，可修改单例实现为按设备标识存储连接（比如用字典保存不同设备的连接实例）

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