使用Python實現批量ping操作方法

在日常的工作中，我們通常會有去探測目標主機是否存活的應用場景，單個的服務器主機可以通過計算機自帶的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('') 
        #print (ip) 
        res = subprocess.call('ping - w1000 - n1 % 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('pleaseinputmodren: ') 
    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()