本文共 6612 字，大约阅读时间需要 22 分钟。
在网络监控领域，了解服务器的网络状态是非常重要的。通过socket模块，我们可以轻松实现从命令行或脚本中获取域名对应的IP地址。以下是详细的实现方案。
我们可以通过socket模块在Python中实现一个简单的ping工具。以下是一个基本的实现示例：
import socketdef ping(host): try: ip = socket.gethostbyname(host) print(f"Host {host} resolved to IP: {ip}") except socket.gaierror: print(f"Could not resolve host: {host}") 当需要批量检查大量服务器的网络状态时，fping是一个非常有用的工具。它能够在瞬间完成对多个IP地址的ping操作，只有当某个IP地址无法连接时，才会逐步增加延迟时间。以下是一个简单的fping脚本示例：
#!/bin/bashrm -f result.txtcat ip_list.txt | fping > result.txt
为了实现自动化，我们可以将IP地址存储在数据库中，并通过脚本定期执行ping检查。以下是一个Python脚本示例：
import subprocessimport osdef check_online_status(ip_list): if os.path.exists('iplist.txt'): os.remove('iplist.txt') with open('iplist.txt', 'a') as f: for ip in ip_list: f.write(f"{ip}\n") p = subprocess.Popen(r'./fping.sh', stdout=subprocess.PIPE) p.stdout.read() result_file = 'result.txt' with open(result_file, 'r') as f: content = f.read().split('\n') for line in content: if not line: continue parts = line.split() ip = parts[0] status = 0 if 'unreachable' in line: status = 1 # 更新数据库状态 cmd = f"update ip_check set status={status} where ip='{ip}'" os.system(cmd)def main(): # 从数据库中获取IP列表 ip_list = mysql('select ip from ip_check') check_online_status(ip_list) print('所有IP状态检查完毕!')if __name__ == '__main__': main() 为了实现自动化，我们可以将脚本设置为计划任务定期运行。以下是如何在Linux上设置计划任务的示例：
chmod +x ping_check.sh*/5 * * * * /path/to/ping_check.sh
以下是完整的Python实现代码，包含了verbose模式和quiet模式：
#!/usr/bin/env python"""A pure Python ping implementation using raw socket.Note that ICMP messages can only be sent from processes running as root.Derived from ping.c distributed in Linux's netkit."""import osimport selectimport socketimport structimport sysimport time# ICMP类型标识符ICMP_ECHO_REQUEST = 8def checksum(source_string): """Calculates the checksum of the given string.""" sum = 0 count_to = (len(source_string) // 2) * 2 for count in range(0, count_to, 2): this = ord(source_string[count + 1]) * 256 + ord(source_string[count]) sum += this sum &= 0xFFFFFFFF # 操作符位宽和 if count_to < len(source_string): sum += ord(source_string[len(source_string) - 1]) sum &= 0xFFFFFFFF sum = (sum >> 16) + (sum & 0xFFFF) sum = sum + (sum >> 16) answer = ~sum answer &= 0xFFFF # 交换字节 answer = (answer >> 8) | (answer << 8 & 0xFF00) return answerdef receive_one_ping(my_socket, id, timeout): """Receives a ping response from the socket.""" time_left = timeout while True: started_select = time.time() what_ready = select.select([my_socket], [], [], time_left) how_long_in_select = time.time() - started_select if not what_ready[0]: return None time_received = time.time() received_packet, addr = my_socket.recvfrom(1024) if not received_packet: continue icmpHeader = received_packet[20:28] type, code, checksum, packet_id, sequence = struct.unpack( "bbHHh", icmpHeader ) if packet_id == id: bytes = struct.calcsize("d") time_sent = struct.unpack("d", received_packet[28:28 + bytes])[0] return time_received - time_sent time_left = time_left - how_long_in_select if time_left <= 0: return Nonedef send_one_ping(my_socket, dest_addr, id, psize): """Sends a ping to the specified address.""" dest_addr = socket.gethostbyname(dest_addr) psize -= 8 # ICMP头部占用8字节 my_checksum = 0 # 创建一个带有0校验和的头部 header = struct.pack("bbHHh", ICMP_ECHO_REQUEST, 0, my_checksum, id, 1) data = (psize - struct.calcsize("d")) * "Q" timestamp = struct.pack("d", time.time()) data = timestamp + data # 计算完整的校验和 my_checksum = checksum(header + data) # 创建最终的头部 header = struct.pack( "bbHHh", ICMP_ECHO_REQUEST, 0, socket.htons(my_checksum), id, 1 ) packet = header + data my_socket.sendto(packet, (dest_addr, 1))def do_one(dest_addr, timeout, psize): """Performs a single ping test and returns the delay.""" icmp = socket.getprotobyname("icmp") try: my_socket = socket.socket(socket.AF_INET, socket.SOCK_RAW, icmp) except socket.error as e: if e.errno == 1: raise socket.error("ICMP messages can only be sent from processes running as root.") raise my_id = os.getpid() & 0xFFFF send_one_ping(my_socket, dest_addr, my_id, psize) my_socket.close() return receive_one_ping(my_socket, my_id, timeout)def verbose_ping(dest_addr, timeout=2, count=4, psize=64): """Sends `count` pings to `dest_addr` and displays the results.""" for i in range(count): print(f"ping {dest_addr} with ...") try: delay = do_one(dest_addr, timeout, psize) except socket.gaierror as e: print(f"failed. (socket error: {e[1]})") break if delay is None: print(f"failed. (timeout within {timeout} seconds)") else: delay *= 1000 print(f"get ping in {delay:.4f}ms") print()def quiet_ping(dest_addr, timeout=2, count=4, psize=64): """Sends `count` pings to `dest_addr` and returns statistics.""" lost = 0 plist = [] for i in range(count): try: delay = do_one(dest_addr, timeout, psize) delay *= 1000 plist.append(delay) except socket.gaierror as e: print(f"failed. (socket error: {e[1]})") break percent_lost = 100 - (len(plist) * 100 // count) if plist: max_rtt = max(plist) avg_rtt = sum(plist) / len(plist) else: max_rtt = None avg_rtt = None print(f"{percent_lost}% packets lost") if max_rtt is not None: print(f"Max RTT: {max_rtt:.2f}ms") print(f"Avg RTT: {avg_rtt:.2f}ms") print(len(plist))if __name__ == '__main__': # verbose_ping("heise.de") # verbose_ping("google.com") # verbose_ping("a-test-url-taht-is-not-available.com") verbose_ping("www.xd.com") print(quiet_ping("www.xd.com", count=10)) 主要功能：
verbose_ping：显示详细的ping结果。quiet_ping：返回丢包率、最大延迟和平均延迟。适用场景：
注意事项：
通过以上方法，我们可以轻松实现网络状态监控，适用于服务器数量较多的场景。
转载地址：http://smafk.baihongyu.com/