如何用pysnmp快速获取多个SNMP OID值？解决PDU查询超时问题

核心方案：使用SNMP GETBULK替代GET请求 #

你的OID属于同一个MIB表的同列（.1.3.6.1.4.1.3808.1.1.3.3.5.1.1.4是CyberPower PDU的端口状态列），SNMPv2c的GETBULK操作专门用于高效批量获取表数据，比多次GET或单次多OIDGET更适配这类场景，能大幅减少请求次数与响应时间，解决超时问题。

pysnmp实现代码 #

from pysnmp.hlapi import (
    SnmpEngine, CommunityData, UdpTransportTarget,
    ContextData, ObjectType, ObjectIdentity, bulkCmd
)

def convert_snmp_type_to_python_type(value):
    # 保留你原有的类型转换逻辑，示例如下：
    if hasattr(value, 'prettyPrint'):
        val_str = value.prettyPrint()
        if val_str.isdigit():
            return int(val_str)
        return val_str
    return value

def get_all_outlet_states(ip_address: str) -> dict[int, int | str]:
    """批量获取1-16端口状态，返回{端口号: 状态值}字典"""
    # 目标列OID：CyberPower PDU端口状态列（不含端口后缀）
    column_oid = ".1.3.6.1.4.1.3808.1.1.3.3.5.1.1.4"
    
    iterator = bulkCmd(
        SnmpEngine(),
        CommunityData("public", mpModel=0),  # mpModel=0对应SNMPv2c
        UdpTransportTarget(transportAddr=(ip_address, 161), timeout=1, retries=0),
        ContextData(),
        0,  # nonRepeaters：0表示所有OID都重复获取
        16, # maxRepetitions：最多返回16条，匹配端口数量
        ObjectType(ObjectIdentity(column_oid)),
        lexicographicMode=False  # 仅返回目标列条目，避免遍历整个MIB树
    )

    outlet_states = {}
    error_indication = None
    error_status = None
    
    for error_indication, error_status, error_index, variable_bindings in iterator:
        if error_indication or error_status:
            break
        
        for var_binding in variable_bindings:
            oid, value = var_binding
            # 从OID末尾提取端口号
            outlet_num = int(str(oid).split('.')[-1])
            outlet_states[outlet_num] = convert_snmp_type_to_python_type(value)
    
    if error_indication:
        raise RuntimeError(str(error_indication))
    elif error_status:
        raise RuntimeError(f"{error_status.prettyPrint()} at {error_index and variable_bindings[int(error_index)-1][0] or '?'}")
    
    # 确保返回1-16端口的完整状态（防止设备返回不全）
    for outlet in range(1, 17):
        outlet_states.setdefault(outlet, None)
    
    return outlet_states

关键优化点 #

• GETBULK适配表结构：针对同列批量查询，GETBULK仅需一次请求即可获取所有端口状态，避免多OID GET的PDU大小限制问题。
• 复用SnmpEngine：若在子进程循环查询，建议在循环外初始化一次SnmpEngine，减少重复初始化的性能开销。
• 参数精准配置：maxRepetitions设为16匹配端口数，timeout设为1秒足够满足低延迟需求，lexicographicMode=False避免无效数据传输。

子进程/线程适配示例 #

为满足1-2Hz的循环查询需求，可将查询逻辑放在独立线程中，避免阻塞GUI主线程：

import threading
import time

def monitor_pdu_states(ip, interval=0.5):
    """每隔0.5秒查询一次端口状态（2Hz频率）"""
    engine = SnmpEngine()  # 复用引擎减少开销
    while True:
        try:
            states = get_all_outlet_states(ip)
            # 此处将states传递给GUI更新（例如通过队列）
            print(states)
        except Exception as e:
            print(f"查询失败: {e}")
        time.sleep(interval)

# 启动监控线程
threading.Thread(target=monitor_pdu_states, args=("192.168.1.100",), daemon=True).start()

替代方案（若pysnmp仍有问题） #

如果bulkCmd仍存在异常，可尝试snmpclitools（基于pysnmp的纯Python跨平台封装），核心逻辑仍为GETBULK操作，代码复杂度略高但兼容性一致。

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