Python多线程Ping网络测试脚本实现与解析
发布时间：2026-10-13| 3人浏览过
本文分享一个基于Python的网络测试工具，通过多线程并发ping指定IP列表，利用xlrd和xlwt库实现Excel数据的读写，并采用subprocess模块解决Windows平台兼容性问题。文章详细解析了代码逻辑、线程处理函数及正则表达式匹配现状，旨在为需要批量检测网络状态的用户提供实用参考。
前言
近期应同学请求，协助开发一款网络测试工具。受工作事务影响，该项目历经一段时间才呈现出相对完整的版本。由于本人日常使用Python频率较低，开发过程主要依赖查阅资料边学边写。
程序的核心逻辑设计如下：
首先读取Excel文件中的IP地址列表，随后利用多线程技术并发执行ping命令以统计各IP的网络参数，最终将统计结果输出至新的Excel文件中。
具体代码实现如下：
#! /usr/bin/env python
# -*- coding: UTF-8 -*-
# File: pingtest_test.py
# Date: 2008-09-28
# Author: Michael Field
# Modified By：intheworld
# Date: 2017-4-17
import sys
import os
import getopt
import commands
import subprocess
import re
import time
import threading
import xlrd
import xlwt
TEST = [
'220.181.57.217',
'166.111.8.28',
'202.114.0.242',
'202.117.0.20',
'202.112.26.34',
'202.203.128.33',
'202.115.64.33',
'202.201.48.2',
'202.114.0.242',
'202.116.160.33',
'202.202.128.33',
]
RESULT={}
def usage():
print "USEAGE:"
print "\t%s -n TEST|excel name [-t times of ping] [-c concurrent number(thread nums)]" %sys.argv[0]
print "\t TEST为简单测试的IP列表"
print "\t-t times 测试次数;默认为1000;"
print "\t-c concurrent number 并行线程数目:默认为10"
print "\t-h|-?, 帮助信息"
print "\t 输出为当前目录文件ping_result.txt 和 ping_result.xls"
print "for example:"
print "\t./ping_test.py -n TEST -t 1 -c 10"
def div_list(ls,n):
if not isinstance(ls,list) or not isinstance(n,int):
return []
ls_len = len(ls)
print 'ls length = %s' %ls_len
if n<=0 or 0==ls_len:
return []
if n > ls_len:
return []
elif n == ls_len:
return [[i] for i in ls]
else:
j = ls_len/n
k = ls_len%n
### j,j,j,...(前面有n-1个j),j+k
#步长j,次数n-1
ls_return = []
for i in xrange(0,(n-1)*j,j):
ls_return.append(ls[i:i+j])
#算上末尾的j+k
ls_return.append(ls[(n-1)*j:])
return ls_return
def pin(IP):
try:
xpin=subprocess.check_output("ping -n 1 -w 100 %s" %IP, shell=True)
except Exception:
xpin = 'empty'
ms = '=[0-9]+ms'.decode("utf8")
print "%s" %ms
print "%s" %xpin
mstime=re.search(ms,xpin)
if not mstime:
MS='timeout'
return MS
else:
MS=mstime.group().split('=')[1]
return MS.strip('ms')
def count(total_count,I):
global RESULT
nowsecond = int(time.time())
nums = 0
oknums = 0
timeout = 0
lostpacket = 0.0
total_ms = 0.0
avgms = 0.0
maxms = -1
while nums < total_count:
nums += 1
MS = pin(I)
print 'pin output = %s' %MS
if MS == 'timeout':
timeout += 1
lostpacket = timeout*100.0 / nums
else:
oknums += 1
total_ms = total_ms + float(MS)
if oknums == 0:
oknums = 1
maxms = float(MS)
avgms = total_ms / oknums
else:
avgms = total_ms / oknums
maxms = max(maxms, float(MS))
RESULT[I] = (I, avgms, maxms, lostpacket)
def thread_func(t, ipList):
if not isinstance(ipList,list):
return
else:
for ip in ipList:
count(t, ip)
def readIpsInFile(excelName):
data = xlrd.open_workbook(excelName)
table = data.sheets()[0]
nrows = table.nrows
print 'nrows %s' %nrows
ips = []
for i in range(nrows):
ips.append(table.cell_value(i, 0))
print table.cell_value(i, 0)
return ips
if __name__ == '__main__':
file = 'ping_result.txt'
times = 10
network = ''
thread_num = 10
args = sys.argv[1:]
try:
(opts, getopts) = getopt.getopt(args, 'n:t:c:h?')
except:
print "\nInvalid command line option detected."
usage()
sys.exit(1)
for opt, arg in opts:
if opt in ('-n'):
network = arg
if opt in ('-h', '-?'):
usage()
sys.exit(0)
if opt in ('-t'):
times = int(arg)
if opt in ('-c'):
thread_num = int(arg)
f = open(file, 'w')
workbook = xlwt.Workbook()
sheet1 = workbook.add_sheet("sheet1", cell_overwrite_ok=True)
if not isinstance(times,int):
usage()
sys.exit(0)
if network not in ['TEST'] and not os.path.exists(os.path.join(os.path.dirname(__file__), network)):
print "The network is wrong or excel file does not exist. please check it."
usage()
sys.exit(0)
else:
if network == 'TEST':
ips = TEST
else:
ips = readIpsInFile(network)
print 'Starting...'
threads = []
nest_list = div_list(ips, thread_num)
loops = range(len(nest_list))
print 'Total %s Threads is working...' %len(nest_list)
for ipList in nest_list:
t = threading.Thread(target=thread_func,args=(times,ipList))
threads.append(t)
for i in loops:
threads[i].start()
for i in loops:
threads[i].join()
it = 0
for line in RESULT:
value = RESULT[line]
sheet1.write(it, 0, line)
sheet1.write(it, 1, str('%.2f'%value[1]))
sheet1.write(it, 2, str('%.2f'%value[2]))
sheet1.write(it, 3, str('%.2f'%value[3]))
it+=1
f.write(line + '\t'+ str('%.2f'%value[1]) + '\t'+ str('%.2f'%value[2]) + '\t'+ str('%.2f'%value[3]) + '\n')
f.close()
workbook.save('ping_result.xls')
print 'Work Done. please check result %s and ping_result.xls.'%file
该段代码参考了现有实现方案，尽管逻辑不算复杂，此处仍对关键部分进行简要说明。
- Excel文件的读写操作依赖于xlrd和xlwt库，主要调用了其基础API接口。
- 多线程并发功能通过threading模块实现，其用法与POSIX标准接口相似。其中thread_func为线程处理函数，接收IP列表作为输入，并在函数内部通过循环逐一处理各个IP地址。
- 此外，考虑到Python的commands模块在Windows环境下存在兼容性问题，代码中改用了subprocess模块来执行系统命令。
截至目前，我对Python字符集的处理机制理解尚浅，因此正则表达式匹配的代码健壮性不足，但当前版本已能勉强运行，未来若有必要再进行优化改进。
总结
以上就是本文的全部内容，希望本文的内容对大家的学习或者工作能带来一定的帮助，如果有疑问大家可以留言交流，谢谢大家对本站的支持。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
