Python多线程批量Ping探测存活主机实战

追风的人

发布时间：2026-10-13 | 3人浏览过

原创

本文介绍如何利用Python的subprocess、threading及queue模块，通过多线程技术实现高效的批量Ping操作。针对传统DOS命令在探测C段主机时速度缓慢的问题，提供完整的Python3代码方案，支持指定IP列表或自动遍历C段，并集成日志记录功能，帮助运维人员快速定位网段内存活主机。

批量Ping探测的背景与方案

在日常运维工作中，探测目标主机是否在线是常见需求。对于单一服务器，使用系统自带的命令即可处理；然而实际业务环境中，往往需要探测整个C段（同一网段）内的存活主机。此时若依赖传统命令行工具，效率过低，无法满足实际工作对速度的要求。因此，通常需要通过脚本实现一键部署与执行。本文主要演示如何使用Python语言，结合多线程技术，来实现高效的批量Ping操作。

环境依赖与库说明

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操作的文章就介绍到这了,更多相关使用Python实现批量ping操作内容请搜索本站以前的文章或继续浏览下面的相关文章希望大家以后多多支持本站！