Python检测UDP端口是否正常通信

本文主要是介绍Python检测UDP端口是否正常通信，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

• 通过python脚本构建客户端，向服务器发送hello的UDP包，然后在跑客户端的机器上tcpdump抓包查看是否能正常收到UDP回包，即可判断UDP服务是否正常。

# -*- coding:utf-8 -*-
#!/usr/bin/python

import sys
import os
import socket
import threading
from threading import _Timer
import datetime
import time
import select

totalTime = 0
sendCnt = 0
failCnt = 0
timeout = 10
timeoutCnt = 0

bufsize = 1024

logFile = 'ping.data'

class RepeatTimer(_Timer):
    def run(self):
        while not self.finished.is_set():
            self.function(*self.args, **self.kwargs)
            self.finished.wait(self.interval)

def GetTime(func):
    def Wrapper(*arg, **kwarg):
        begin = datetime.datetime.now()
        func(*arg, **kwarg)
        end = datetime.datetime.now()

        costMs = (end - begin).seconds * 1000 + (end - begin).microseconds / 1000

        global totalTime

        totalTime = totalTime + costMs

        print "costMs=%dms, totalTime=%dms" % (costMs, totalTime)

    return Wrapper

def SetTimeOut(seconds):
    def Decorator(func):
        def Wrapper(*arg, **kwarg):
            t = threading.Thread(target=func, args=arg, kwargs=kwarg)
            t.setDaemon(True)
            t.start()
            t.join(seconds)

            if t.isAlive():
                raise Exception('function timeout')

        return Wrapper
    return Decorator


@SetTimeOut(timeout)
def RecvData(handle):

    ready = select.select([handle], [], [], timeout)

    if ready[0]:
        recvMsg = handle.recv(bufsize).decode('utf-8')

        if recvMsg == 'hello':
            return True
        else:
            return False

    else:
        print "function timeout: %ds" % timeout
        return False


@GetTime
def PingUdpPort(ip, port, file):
    global sendCnt
    global failCnt
    global timeoutCnt
    sendCnt = sendCnt + 1
    data = 'hello'

    handle = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    handle.setblocking(0)

    handle.sendto(data, (ip, port))

    try:
        state = RecvData(handle)

        if state is False:
            failCnt = failCnt + 1

    #超时，目前设置的是10秒        
    except Exception, e:
        timeoutCnt = timeoutCnt + 1
        print "Run exception", e

    WritePingLog(file)

def WritePingLog(file):
    if totalTime is not 0 and sendCnt > failCnt:
        avgCost = totalTime / (sendCnt - failCnt)
    else:
        avgCost = 0

    file.write("Ping count=%d, fail count=%d, timeoutMsg count=%d, avgCost=%ds\n" % (sendCnt, failCnt, timeoutCnt, avgCost))
    file.flush()

if __name__ == '__main__':
    if len(sys.argv) < 3:
        print "Invalid param count, should be python %s ip port interval" % os.path.basename(__file__)
        exit(-1)

    ip = sys.argv[1]
    port = int(sys.argv[2])
    interval = 2.0
    
    print "argv=", sys.argv

    with open(logFile, 'w') as file:

        testTimer = RepeatTimer(interval, PingUdpPort, args=[ip, port, file])

        testTimer.run()
    # PingUdpPort(handle, ip, port)

• 测试如下

[root@test ~]# python ping_echosvr.py x.x.x.x x
argv= ['ping_echosvr.py', 'x.x.x.x', 'x']
costMs=63ms, totalTime=63ms
costMs=63ms, totalTime=126ms

• 抓包结果如下

[root@test ~]# tcpdump -i eth0 udp and src net x.x.x.x
tcpdump: verbose output suppressed, use -v or -vv for full protocol decode
listening on eth0, link-type EN10MB (Ethernet), capture size 65535 bytes
14:22:05.420606 IP x.x.x.x > vip.43751: UDP, length 5
14:22:07.486959 IP x.x.x.x > vip.44752: UDP, length 5

• 如果抓包正常，说明UDP服务能正常通信，如果偶尔没抓到回包也是正常的，毕竟UDP是无连接的，可靠性较差。

这篇关于Python检测UDP端口是否正常通信的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！