前一篇学习了获取本地 IP,这一篇学习如何批量检测网络中的设备是否在线。ping 是网络运维中最基础的诊断工具。
本篇使用 subprocess 调用系统 ping 命令,结合 threading 多线程实现高效的并发检测。
安装依赖
💡 提示
本篇只需要 Python 内置模块,无需额外安装。
方案一:单线程 ping
最基础的方式,直接调用系统 ping 命令:
Python 代码
import subprocessimport platformdef ping_host(host, count=4, timeout=2): try: if platform.system() == 'Windows': cmd = ['ping', '-n', str(count), '-w', str(timeout * 1000), host] else: cmd = ['ping', '-c', str(count), '-W', str(timeout * 1000), host] result = subprocess.run(cmd, capture_output=True, timeout=timeout + 5, text=True) if result.returncode == 0: output = result.stdout if 'Average' in output: avg_time = output.split('Average = ')[1].split('ms')[0] return True, f"{avg_time}ms" elif 'min/avg/max' in output: avg_time = output.split('min/avg/max/')[1].split(' ')[0] return True, f"{avg_time}ms" else: return True, "在线" else: return False, "无响应" except subprocess.TimeoutExpired: return False, "超时" except Exception as e: return False, str(e)if __name__ == '__main__': hosts = ['8.8.8.8', '192.168.1.1', '10.0.0.1'] for host in hosts: online, time_info = ping_host(host) status = "在线" if online else "离线" print(f"{host}: {status} ({time_info})")
方案二:多线程并发 ping
单线程太慢,用多线程加速。检测 254 个 IP 从 10 分钟降低到 30 秒。
Python 代码
import subprocessimport threadingimport platformfrom datetime import datetimeclass PingScanner: def __init__(self, max_threads=50): self.max_threads = max_threads self.results = [] self.lock = threading.Lock() def ping_host(self, host): try: if platform.system() == 'Windows': cmd = ['ping', '-n', '1', '-w', '1000', host] else: cmd = ['ping', '-c', '1', '-W', '1000', host] result = subprocess.run(cmd, capture_output=True, timeout=3, text=True) online = result.returncode == 0 with self.lock: self.results.append({'host': host, 'online': online}) return online except: with self.lock: self.results.append({'host': host, 'online': False}) return False def scan_network(self, network_prefix, start=1, end=254): threads = [] for i in range(start, end + 1): host = f"{network_prefix}.{i}" while threading.active_count() > self.max_threads: threading.Event().wait(0.1) t = threading.Thread(target=self.ping_host, args=(host,)) t.daemon = True t.start() threads.append(t) for t in threads: t.join() return [r['host'] for r in self.results if r['online']]if __name__ == '__main__': scanner = PingScanner(max_threads=50) online_hosts = scanner.scan_network('192.168.1', start=1, end=254) print(f"在线主机数: {len(online_hosts)}") for host in sorted(online_hosts): print(f" - {host}")
性能对比
📊 性能提升
单线程检测 254 个 IP 需要约 10 分钟,多线程(50 个线程)仅需 30 秒。
注意事项
- 权限要求:Linux 下 ping 可能需要 root 权限
- 网络礼仪
- 防火墙
- 性能优化