- android - 多次调用 OnPrimaryClipChangedListener
- android - 无法更新 RecyclerView 中的 TextView 字段
- android.database.CursorIndexOutOfBoundsException : Index 0 requested, 光标大小为 0
- android - 使用 AppCompat 时,我们是否需要明确指定其 UI 组件(Spinner、EditText)颜色
如果我想并发执行pysnmp
带有 cmdgen.CommandGenerator().nextCmd()
的线程,是 pysnmp
线程安全?
当我测试它时,我没有看到表明有问题的问题;但是,我没有运行 pysnmp
之前使用线程,这些查询的结果将用于做出业务决策,所以我想得到一个明确的答案 pysnmp
的线程安全。
from threading import Thread
import time
# See SNMP.py module at the bottom of the question
from SNMP import v2c
class SNMP_Walk_Thread(Thread):
def __init__(self, address='127.0.0.1', oid='sysDescr'):
Thread.__init__(self)
self.address = address
self.oid = oid
self.results = None
def run(self):
self.snmp = v2c(self.address)
self.results = self.snmp.walk(self.oid)
if __name__=='__main__':
managers = set()
finished = set()
start = time.time()
for addr in ['172.25.116.20', '172.25.116.21', '172.25.116.22',
'172.25.116.5']:
print "Calling %s(address=%s)" % ('SNMP_Walk_Thread', addr)
poll = SNMP_Walk_Thread(address=addr)
poll.start(); managers.add(poll)
while (len(finished)<len(managers)):
for instance in managers.difference(finished):
instance.join(0.5)
if not instance.is_alive():
finished.add(instance)
print "RESULT:", instance.results
print " Execution time: %0.3f seconds" % ((time.time()-start))
"""SNMP.py (Updated to include single CommandGenerator() instances per thread, as
suggested in Ilya's answer"""
from collections import namedtuple as NT
from datetime import datetime
import string
import re
from pysnmp.entity.rfc3413.oneliner import cmdgen
from pysnmp.smi import builder, view, error
from numpy import int64, float64
# NOTE!!!
# It is best to install the pysnmp-mibs package from pypi... this makes
# a lot of symbolic MIB names "just work"
# See this link below for many oneliner examples...
# http://pysnmp.sourceforge.net/examples/4.x/v3arch/oneliner/index.html
class v2c(object):
"""Build an SNMPv2c manager object"""
def __init__(self, ipaddr=None, device=None, community='Public',
retries=3, timeout=9):
self.device = device
self.ipaddr = ipaddr
self.community = community
self.SNMPObject = NT('SNMPObject', ['modName', 'datetime', 'symName',
'index', 'value'])
self.SNMPIndexed = NT('SNMPIndexed', ['modName', 'datetime', 'symName',
'index', 'value'])
self.query_timeout = float(timeout)/int(retries)
self.query_retries = int(retries)
self._index = None
self.cmdGen = cmdgen.CommandGenerator()
#mibBuilder = builder.MibBuilder()
#mibPath = mibBuilder.getMibPath()+('/opt/python/Models/Network/MIBs',)
#mibBuilder.setMibPath(*mibPath)
#mibBuilder.loadModules(
# 'RFC-1213',
# )
#mibView = view.MibViewController(mibBuilder)
def index(self, oid=None):
"""Build an SNMP Manager index to reference in get or walk operations. First v2c.index('ifName'). Then, v2c.get_index('ifHCInOctets', 'eth0') or v2c.walk_index('ifHCInOctets'). Instead of referencing a numerical index, the index will refer to the value that was indexed."""
self._index = dict()
self._intfobj = dict()
snmpidx = self.walk(oid=oid)
for ii in snmpidx:
## the dicts below are keyed by the SNMP index number
# value below is the text string of the intf name
self._index[ii.index] = ii.value
# value below is the intf object
if not (self.device is None):
self._intfobj[ii.index] = self.device.find_match_intf(ii.value,
enforce_format=False)
def walk_index(self, oid=None):
"""Example usage, first index with v2c.index('ifName'), then v2c.get_index('ifHCInOctets', 'eth0')"""
if not (self._index is None):
tmp = list()
snmpvals = self.walk(oid=oid)
for idx, ii in enumerate(snmpvals):
tmp.append([ii.modName, datetime.now(), ii.symName,
self._index[ii.index], ii.value])
return map(self.SNMPIndexed._make, tmp)
else:
raise ValueError, "Must populate with SNMP.v2c.index() first"
def walk(self, oid=None):
if isinstance(self._format(oid), tuple):
errorIndication, errorStatus, errorIndex, \
varBindTable = self.cmdGen.nextCmd(
cmdgen.CommunityData('test-agent', self.community),
cmdgen.UdpTransportTarget((self.ipaddr, 161),
retries=self.query_retries,
timeout=self.query_timeout),
self._format(oid),
)
# Parsing only for now... no return value...
self._parse(errorIndication, errorStatus, errorIndex, varBindTable)
elif isinstance(oid, str):
errorIndication, errorStatus, errorIndex, \
varBindTable = self.cmdGen.nextCmd(
# SNMP v2
cmdgen.CommunityData('test-agent', self.community),
# Transport
cmdgen.UdpTransportTarget((self.ipaddr, 161)),
(('', oid),),
)
return self._parse_resolve(errorIndication, errorStatus,
errorIndex, varBindTable)
else:
raise ValueError, "Unknown oid format: %s" % oid
def get_index(self, oid=None, index=None):
"""In this case, index should be similar to the values you indexed from... i.e. if you index with ifName, get_index('ifHCInOctets', 'eth0')"""
if not (self._index is None) and isinstance(index, str):
# Map the interface name provided in index to an ifName index...
snmpvals = None
for idx, value in self._index.items():
if index == value:
# if there is an exact match between the text index and the
# snmp index value...
snmpvals = self.get(oid=oid, index=idx)
break
else:
# TRY mapping the provided text index into an interface obj
_intfobj = self.device.find_match_intf(index)
if not (_intfobj is None):
for key, val in self._intfobj.items():
if (val==_intfobj):
snmpvals = self.get(oid=oid, index=key)
break
# Ensure we only parse a valid response...
if not (snmpvals is None):
tmp = [snmpvals.modName, datetime.now(), snmpvals.symName,
self._index[snmpvals.index], snmpvals.value]
return self.SNMPIndexed._make(tmp)
elif not isinstance(index, str):
raise ValueError, "index must be a string value"
else:
raise ValueError, "Must populate with SNMP.v2c.index() first"
def get(self, oid=None, index=None):
if isinstance(self._format(oid), tuple):
errorIndication, errorStatus, errorIndex, \
varBindTable = self.cmdGen.getCmd(
cmdgen.CommunityData('test-agent', self.community),
cmdgen.UdpTransportTarget((self.ipaddr, 161),
retries=self.query_retries,
timeout=self.query_timeout),
self._format(oid),
)
# Parsing only for now... no return value...
self._parse(errorIndication, errorStatus, errorIndex, varBindTable)
elif isinstance(oid, str) and isinstance(index, int):
errorIndication, errorStatus, errorIndex, \
varBindTable = self.cmdGen.getCmd(
# SNMP v2
cmdgen.CommunityData('test-agent', self.community),
# Transport
cmdgen.UdpTransportTarget((self.ipaddr, 161)),
(('', oid), index),
)
return self._parse_resolve(errorIndication, errorStatus,
errorIndex, [varBindTable])[0]
else:
raise ValueError, "Unknown oid format: %s" % oid
def bulkwalk(self, oid=None):
"""SNMP bulkwalk a device. NOTE: This often is faster, but does not work as well as a simple SNMP walk"""
if isinstance(self._format(oid), tuple):
errorIndication, errorStatus, errorIndex, varBindTable = self.cmdGen.bulkCmd(
cmdgen.CommunityData('test-agent', self.community),
cmdgen.UdpTransportTarget((self.ipaddr, 161),
retries=self.query_retries,
timeout=self.query_timeout),
0,
25,
self._format(oid),
)
return self._parse(errorIndication, errorStatus,
errorIndex, varBindTable)
elif isinstance(oid, str):
errorIndication, errorStatus, errorIndex, varBindTable = self.cmdGen.bulkCmd(
cmdgen.CommunityData('test-agent', self.community),
cmdgen.UdpTransportTarget((self.ipaddr, 161),
retries=self.query_retries,
timeout=self.query_timeout),
0,
25,
(('', oid),),
)
return self._parse_resolve(errorIndication, errorStatus,
errorIndex, varBindTable)
else:
raise ValueError, "Unknown oid format: %s" % oid
def _parse_resolve(self, errorIndication=None, errorStatus=None,
errorIndex=None, varBindTable=None):
"""Parse MIB walks and resolve into MIB names"""
retval = list()
if errorIndication:
print errorIndication
else:
if errorStatus:
print '%s at %s\n' % (
errorStatus.prettyPrint(),
varBindTable[-1][int(errorIndex)-1]
)
else:
for varBindTableRow in varBindTable:
for oid, val in varBindTableRow:
(symName, modName), indices = cmdgen.mibvar.oidToMibName(
self.cmdGen.mibViewController, oid
)
val = cmdgen.mibvar.cloneFromMibValue(
self.cmdGen.mibViewController, modName, symName,
val)
# Try to parse the index as an int first,
# then as a string
try:
index = int(string.join(map(lambda v: v.prettyPrint(), indices), '.'))
except ValueError:
index = str(string.join(map(lambda v: v.prettyPrint(), indices), '.'))
# Re-format values as float or integer, if possible...
tmp = val.prettyPrint()
if re.search(r"""^\s*\d+\s*$""", tmp):
value = int64(tmp)
elif re.search(r"""^\s*\d+\.\d+\s*$""", tmp):
value = float64(tmp)
else:
value = tmp
retval.append(self.SNMPObject._make([modName,
datetime.now(), symName, index, value]))
return retval
def _parse(self, errorIndication, errorStatus, errorIndex,
varBindTable):
if errorIndication:
print errorIndication
else:
if errorStatus:
print '%s at %s\n' % (
errorStatus.prettyPrint(),
errorIndex and varBindTable[-1][int(errorIndex)-1] or '?'
)
else:
for varBindTableRow in varBindTable:
for name, val in varBindTableRow:
print '%s = %s' % (name.prettyPrint(), val.prettyPrint())
def _format(self, oid):
"""Format a numerical OID in the form of 1.3.4.1.2.1 into a tuple"""
if isinstance(oid, str):
if re.search('(\d+\.)+\d+', oid):
tmp = list()
for ii in oid.split('.'):
tmp.append(int(ii))
return tuple(tmp)
else:
return oid
最佳答案
如果您在各自的线程中运行专用的 CommandGenerator 实例,pysnmp 应该是 MT 安全的。最新的 pysnmp 版本 (4.2.2) 有相关修复。
我会考虑对您的代码进行以下更改:
关于python - pysnmp 是线程安全的吗?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/10381963/
我正在 Windows XP 机器上运行 Python 程序。当我运行该程序时,出现以下错误: File "C:\Python27\lib\pysnmp\smi\builder.pyt, line 2
我在 ubuntu 15.10 中安装了软件包及其依赖模块,但它仍然显示错误 从 pysnmp.hlapi 导入 *ImportError: 没有名为 hlapi 的模块 我尝试安装它sudo set
我想开发简单的 pysnmp 命令响应器来监听我的测试仪器设备,获取来自设备的请求,并在我的主机 PC(Ubuntu)中用我自己的命令(Telnet 命令)替换它们并将其发送到评估板。过去HOST上也
我正在尝试利用 pysnmp 中的 setCmd() 方法来设置变量。我在设置特定对象标识时遇到问题,因为 pysnmp 似乎将“.0”附加到我想要设置的对象标识。为什么会发生这种情况? 我得到的输出
为了获得更好的性能,我必须将我的 bash 脚本迁移到 python 脚本...所以我开始使用 pysnmp,但我遇到了有关输出格式的问题... 您将在下面找到 netsnmp 请求: snmpwal
如果我想并发执行pysnmp带有 cmdgen.CommandGenerator().nextCmd() 的线程,是 pysnmp线程安全? 当我测试它时,我没有看到表明有问题的问题;但是,我没有运行
我正在使用以下简单脚本: from pysnmp.entity.rfc3413.oneliner import cmdgen errorIndication, errorStatus, errorIn
我正在使用 PySNMP 作为 SNMP 代理来实现一个表。 我遵循使用自动生成的 TRS-MIB.py 文件和包含实现的手动编写的 __TRS-MIB.py 文件的(PySNMP 源)架构。 我已经
我刚开始使用 PYSNMP 协议(protocol),我正在尝试通过 SNMP 协议(protocol)从交换机断开和连接端口。我已经有一个与计算机通信的交换机,在本例中将是一个树莓派 3,但我需要一
我需要通过简单的 snmpget 查询询问 4000 台主机。我使用带线程的 netsnmp 和带扭曲的 pynetsnmp,它工作得非常快(不到 1 分钟)。我尝试将 pysnmp 与 AsyncC
我目前有一个脚本,可以使用 PySNMP 轮询多个设备上的多个 oid。它从文件中读取主机列表,对于某些主机需要轮询 3 或 4 个 oid,目前它是按顺序执行此操作,因此为了提高效率,我想做一个 g
我正在尝试将生成的 PYSNMP 数据存储到 mysql 数据库中。 生成的数据是元组还是列表——我不确定。 我得到的输出是 9016,但是当我希望将其保存到数据库时,它类似于 (OID(1.3.6.
我是Python新手。我正在尝试了解 pysnmp 的用法。 我尝试了以下方法: import asyncio from pysnmp.hlapi.asyncio import * from pysn
你好,我正在尝试使用 pysnmp 文档中的这段代码来监听陷阱: from pysnmp.carrier.asynsock.dispatch import AsynsockDispatcher fro
面对 pysnmp 代理返回值的奇怪问题。 PYSNMP 代理接收字符串格式为 "0a0a0a0a0a0a0a0a0a" . 在 MIB 中,OID 表示为 HEXA STRING。 当尝试使用 rf
我有一个想要返回的 SNMP OID 列表,但我无法用变量替换命令中的 OID。我什至尝试将变量包装在 eval 中,并与 ObjectType 和 ObjectIdentity 结合使用,但没有成功
我正在使用 python 2.7 并尝试使用 pysnmp 捕获 SNMP 陷阱。我正在使用 http://pysnmp.sourceforge.net/examples/current/v1arch
我正在尝试制作一个可以在 Pysnmp 中使用 IPv6 的程序,并且我已经从 google 等阅读了很多内容。 而且我总是发现人们使用Udp6SocketTransport。但问题是这个类不能在所有
我正在使用以下代码: import pysnmp from pysnmp.entity.rfc3413.oneliner import cmdgen errorIndication, errorSta
我正在尝试通过 SNMP 从 ucs-6100 获得答案。 通过控制台:(模拟我的请求) >>> snmpget -v3 -l authPriv -u usr-sha-aes -A authkey1
我是一名优秀的程序员,十分优秀!