Python多线程判断IP可达性的实现与优化

Evan

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

原创

本文详解如何使用Python的subprocess模块调用ping命令判断IP是否可达，并结合threading实现多线程并发检测。针对基础版本存在的线程竞争问题，进一步介绍利用Queue队列进行优化的方案，通过FIFO机制提升并发处理的稳定性与效率。

在Python中，可以通过subprocess模块来调用Shell命令。

关于threading模块的使用方式，可以通过创建线程对象来执行特定任务。

from __future__ import print_function
import subprocess
import threading

def is_reachable(ip):
  if subprocess.call(["ping", "-c", "2", ip])==0:#只发送两个ECHO_REQUEST包
    print("{0} is alive.".format(ip))
  else:
    print("{0} is unalive".format(ip))
if __name__ == "__main__":
  ips = ["www.baidu.com","192.168.0.1"]
  threads = []
  for ip in ips:
    thr = threading.Thread(target=is_reachable, args=(ip,))#参数必须为tuple形式
    thr.start()#启动
    threads.append(thr)
  for thr in threads:
    thr.join()

改进方案：引入Queue队列进行优化（采用FIFO先进先出模式）。

from __future__ import print_function
import subprocess
import threading
from Queue import Queue
from Queue import Empty

def call_ping(ip):
  if subprocess.call(["ping", "-c", "2", ip])==0:
    print("{0} is reachable".format(ip))
  else:
    print("{0} is unreachable".format(ip))


def is_reachable(q):
  try:
    while True:
      ip = q.get_nowait()#当队列为空，不等待
      call_ping(ip)
  except Empty:
    pass


def main():
  q = Queue()
  args = ["www.baidu.com", "www.sohu.com", "192.168.0.1"]
  for arg in args:
    q.put(arg)

  threads = []
  for i in range(10):
    thr = threading.Thread(target=is_reachable, args=(q,))
    thr.start()
    threads.append(thr)
  for thr in threads:
    thr.join()

if __name__ == "__main__":
  main()

以上关于Python判断IP是否可达的实例详解内容分享完毕，希望能为读者提供参考，并欢迎大家继续支持本站。