本文主要向大家介紹了Python語言 實現端口連通性檢測，通過具體的內容向大家展示，希望對大家學習Python語言有所幫助。

# -*- coding: utf-8 -*-

#!/bin/env python

#AUTHOR:karl

#DATE:2018-1-19

#VERSION:V1.0

######################

import time

import os

import paramiko

import datetime

import sys

import MySQLdb

import threading

#######################################

#首次插入數據是REG=0，更新數據REG為其他

#######################################

REG=1

private_key = paramiko.RSAKey.from_private_key_file('/home/appdeploy/.ssh/id_rsa')

def Data_mysql(info):

try:

string=info

Pip=string.split(" ")[0]

Sip=string.split(" ")[1]

Dip=string.split(" ")[2]

Port=string.split(" ")[3]

Result=string.split(" ")[4].replace('\n','')

con = MySQLdb.connect(host='localhost',user="root",passwd="******",db="zabbix",port=3306,charset="utf8")

dt = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

cursor=con.cursor()

valuse=(dt,Result,Pip,Sip,Dip,Port)

if REG == 0:

sql="""INSERT INTO Balant_telnet

(create_time,telnet_result,telnet_physicalip,telnet_sourceip,telnet_desip,telnet_port)

VALUES (%s,%s,%s,%s,%s,%s) """

result=cursor.execute(sql,valuse)

else:

sql="""UPDATE Balant_telnet set

create_time='%s',telnet_result=%s where telnet_physicalip='%s' and

telnet_sourceip='%s' and telnet_desip='%s' and telnet_port='%s';"""

%(valuse)

result=cursor.execute(sql)

cursor.close()

con.commit()

con.close()

except MySQLdb.Error, e:

print "Error %d: %s" % (e.args[0], e.args[1])

sys.exit(1)

def check_port(value,content):

ager={}

ager[value]=content

ssh=paramiko.SSHClient()

try:

ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

ssh.connect(value,port=22,username='appdeploy',pkey=private_key,timeout=5)

cmd="sh /home/appdeploy/monitor.sh {value}".format(value=ager[ager.keys()[index]])

stdin,stdout,stderr = ssh.exec_command(cmd)

for file_d in stdout.readlines():

Data_mysql(file_d)

except:

result=cmd+','+'failed'+'\n'

ssh.close()

###################################

# 對配置文件進行參數解析

###################################

def get_parameter():

ager={}

value=[]

with open("monitor_config") as context:

for line in context:

while not "," in list(line):

host=line.strip()[1:-1]

value=[]

break

line=line.strip().split(",")

value=value+line

ager[str(host)]=value

return ager

if __name__ == '__main__':

threads=[]

parm=get_parameter()

files=range(len(parm.keys()))

for index,content in parm.items():

t = threading.Thread(target=check_port,args=(index,content))

threads.append(t)

for index in files:

threads[index].start()

for index in files:

threads[index].join()

配置文件：

[10.117.194.23]

10.117.194.77,10.116.41.82,9920

10.117.194.77,10.116.47.12,1080

10.117.194.77,10.116.45.56,1081

[10.117.194.24]

10.117.194.78,10.116.41.82,9920

10.117.194.78,10.116.47.12,1080

10.117.194.78,10.116.45.56,1081

10.117.194.78,10.116.145.33,8001

【】里面的IP 是物理IP，下面分別對應著應用的源IP，目標IP和測試端口號

agent 直接用shell 實現.

數據庫中效果

在zabbix 中 加入對表中數據的監控。在zabbix 的mysql 模板中加入Balant 的監控項，也可以加入圖像，我們最后的結果是要展示在grafana 上進行展示。為1就說明應用端口是正常運行。

本文由職坐標整理并發布，希望對同學們學習Python有所幫助，更多內容請關注職坐標編程語言Python頻道！

總結

以上是生活随笔為你收集整理的python 测试端口连通_Python语言 实现端口连通性检测的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。