内容简介：在日常的工作中，我们通常会有去探测目标主机是否存活的应用场景，单个的服务器主机可以通过计算机自带的DOS命令来执行，但是业务的存在往往不是单个存在的，通常都是需要去探测C段的主机（同一个网段下的存活主机），这样使用DOS来进行操作是不可取，探测的速度太慢了，不满足实际需要。一般批量的操作需要使用脚本进行一键部署执行，本文主要通过使用Python语言来实现批量ping的操作（使用多线程实现Python批量处理）Python版本 ：Python3使用的第三方库：subprocess, logging, thre
在日常的工作中，我们通常会有去探测目标主机是否存活的应用场景，单个的服务器主机可以通过计算机自带的DOS命令来执行，但是业务的存在往往不是单个存在的，通常都是需要去探测C段的主机（同一个网段下的存活主机），这样使用DOS来进行操作是不可取，探测的速度太慢了，不满足实际需要。一般批量的操作需要使用脚本进行一键部署执行，本文主要通过使用 Python 语言来实现批量ping的操作（使用多线程实现Python批量处理）
Python版本 ：Python3
使用的第三方库：subprocess, logging, threading, queue
日志导出模块功能：
def set_logging_format():
logging.basicConfig(level=logging.INFO,
format='%(message)s',
filename='ping_host.log',
filemode='w'
)
console = logging.StreamHandler()
console.setLevel(logging.INFO)
formatter = logging.Formatter('%(message)s')
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)
多线程实现批量操作：
threads = []
THREAD_NUM = 20
user_iput = input('please input modren: ')
if user_iput == 'addr':
IP_L = IP_LIST
if user_iput == 'file':
IP_L = IP_QUEUE
for i in range (THREAD_NUM):
t = threading.Thread(target = ping_IP,args = (IP_L,))
threads.append(t)
for i in range (THREAD_NUM):
threads[i].start()
for i in range (THREAD_NUM):
threads[i].join()
完整代码部分：
import subprocess
import logging
import datetime
import time
import threading
from queue import Queue
import sys
# 实现日志导出
def set_logging_format():
logging.basicConfig(level=logging.INFO,
format='%(message)s',
filename='ping_host.log',
filemode='w'
)
console = logging.StreamHandler()
console.setLevel(logging.INFO)
formatter = logging.Formatter('%(message)s')
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)
# 将需要ping 连接的IP加入队列
def insert_ip_queue(ip_list_path):
IP_QUEUE = Queue()
with open (ip_list_path,'r') as f:
for ip in f.readlines():
IP_QUEUE.put(ip)
f.close()
return IP_QUEUE
def IP_list ():
ip_list = Queue()
for i in range (1,255):
ip = '192.168.1.' + str(i)
ip_list.put(ip)
return ip_list
# print (IP_list())
#print (IP_list())
#定义 ping 函数
def ping_IP (IP_QUEUE):
while not IP_QUEUE.empty():
ip = IP_QUEUE.get().strip('\n')
#print (ip)
res = subprocess.call('ping -w 1000 -n 1 %s' % ip , stdout=subprocess.PIPE,shell=True)
#print (res)
if res == 0:
h =subprocess.getoutput('ping' + ' ' + ip)
#print (h)
if 'TTL=' in h:
res = ('网络可以正常连通平均延时 = %s' % h.split('平均 = ')[1])
else:
res = '网络连接失败!'
today = datetime.datetime.now().strftime("%Y - %m - %d %H : %M : %S")
logging.info("%s IP = %s %s" % (today,ip,res))
def main ():
set_logging_format()
ip_list_path = './hostip.txt'
IP_QUEUE = insert_ip_queue(ip_list_path)
IP_LIST = IP_list()
threads = []
THREAD_NUM = 20
user_iput = input('please input modren: ')
if user_iput == 'addr':
IP_L = IP_LIST
if user_iput == 'file':
IP_L = IP_QUEUE
for i in range (THREAD_NUM):
t = threading.Thread(target = ping_IP,args = (IP_L,))
threads.append(t)
for i in range (THREAD_NUM):
threads[i].start()
for i in range (THREAD_NUM):
threads[i].join()
if __name__ == '__main__':
main()
以上所述就是小编给大家介绍的《使用Python实现批量ping操作》，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持！
猜你喜欢:
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
Usability for the Web
Tom Brinck、Darren Gergle、Scott D. Wood / Morgan Kaufmann / 2001-10-15 / USD 65.95
Every stage in the design of a new web site is an opportunity to meet or miss deadlines and budgetary goals. Every stage is an opportunity to boost or undercut the site's usability. Thi......一起来看看 《Usability for the Web》 这本书的介绍吧!
