4. 套接字与连接状态监控

网络通了没有？延迟高不高？丢包严不严重？

做监控的人，每天都要回答这三个问题。而回答它们最经典的工具，就是 Ping。Ping 背后依赖的协议叫 ICMP。今天我们就把它彻底讲透，然后手写一个能批量 Ping、能算延迟统计的 Python 脚本。

4.1 ICMP 协议：Ping 的底层语言

ICMP，全称 Internet Control Message Protocol，互联网控制报文协议。它不是用来传数据的，而是用来传「控制消息」和「错误报告」的。

说白了，IP 协议只管把包扔出去，至于扔没扔到、扔错了没有，它不管。ICMP 就是那个负责反馈的家伙。

4.1.1 报文结构

一个 ICMP 报文长这样：

字段	长度	说明
Type	1 字节	报文类型（8=请求，0=应答，3=不可达，11=超时）
Code	1 字节	类型细分（如 Type 3 下 Code 0=网络不可达，Code 1=主机不可达）
Checksum	2 字节	校验和，验证报文完整性
Identifier	2 字节	标识符，用于匹配请求与应答
Sequence Number	2 字节	序列号，用于检测丢包和乱序
Payload	可变	数据负载，通常填充时间戳用于计算 RTT

我记得刚入行时，有一次线上反馈说某台服务器「偶尔连不上」。我抓包一看，ICMP Type 3 Code 1 频繁出现——主机不可达。后来发现是 ARP 表项老化导致的。你看，懂协议细节，定位问题就是快。

4.1.2 Ping 的工作流程

1. 源主机构造 ICMP Echo Request（Type=8），填入 Identifier 和 Sequence Number，再打上发送时间戳。
2. IP 层封装后发往目标。
3. 目标收到后，交换 Type 和 Code，原封不动把 Payload 塞回 Echo Reply（Type=0）。
4. 源主机收到 Reply，用当前时间减去 Payload 里的时间戳，得到 RTT。
5. 如果超时没收到，算丢包。

4.2 Python 实现 Ping：用 ping3 库

自己从零写一个 Ping 程序？可以，但要处理 raw socket、校验和计算、超时管理，代码量不小。生产环境我建议直接用 ping3 这个库，封装得干净，跨平台也稳。

4.2.1 安装

pip install ping3

4.2.2 单点 Ping 示例

from ping3 import ping

# 默认超时 4 秒
rtt = ping('8.8.8.8')
if rtt is None:
    print('目标不可达或超时')
else:
    print(f'RTT: {rtt * 1000:.2f} ms')

返回的 rtt 单位是秒。乘以 1000 转成毫秒，更符合我们日常习惯。

4.2.3 常用参数

参数	默认值	说明
timeout	4	超时时间（秒），超过此值算丢包
count	1	发送次数，用于多次 Ping 取平均
size	56	Payload 大小（字节），默认 56 字节
ttl	64	生存时间，防止路由环路

4.3 批量 Ping 与延迟统计

监控场景下，我们通常不是 Ping 一台机器，而是 Ping 一个网段或一组 IP。比如公司有 200 台服务器，每 30 秒 Ping 一轮，看哪些延迟高、哪些丢包。

4.3.1 批量 Ping 脚本

import time
from ping3 import ping
from concurrent.futures import ThreadPoolExecutor, as_completed

def ping_host(host, timeout=2):
    rtt = ping(host, timeout=timeout)
    if rtt is None:
        return host, None
    return host, round(rtt * 1000, 2)

hosts = ['192.168.1.1', '192.168.1.2', '8.8.8.8', '114.114.114.114']

# 用线程池并发 Ping，大幅提升速度
with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {executor.submit(ping_host, h): h for h in hosts}
    results = {}
    for future in as_completed(futures):
        host, rtt = future.result()
        results[host] = rtt

for host, rtt in results.items():
    if rtt is None:
        print(f'{host}: 超时/不可达')
    else:
        print(f'{host}: {rtt} ms')

为什么用线程池？因为 Ping 是 I/O 密集型操作，等待网络响应的过程中 CPU 是空闲的。串行 Ping 200 台，每台等 2 秒，一轮就要 400 秒。并发 10 个线程，40 秒搞定。你想想看，监控系统里时间就是金钱。

4.3.2 延迟统计与告警

光拿到 RTT 还不够，我们要算几个关键指标：

• 平均延迟：所有成功 RTT 的算术平均。
• 最大/最小延迟：一眼看出波动范围。
• 丢包率：超时次数 / 总次数 × 100%。
• 抖动（Jitter）：相邻 RTT 差值的绝对值平均。抖动大说明网络不稳定。

def ping_stat(host, count=10, timeout=2):
    rtts = []
    lost = 0
    for _ in range(count):
        rtt = ping(host, timeout=timeout)
        if rtt is None:
            lost += 1
        else:
            rtts.append(rtt * 1000)
        time.sleep(0.1)  # 避免 Ping 太快被限流
    
    total = count
    loss_rate = lost / total * 100
    if rtts:
        avg_rtt = sum(rtts) / len(rtts)
        max_rtt = max(rtts)
        min_rtt = min(rtts)
        # 计算抖动
        jitter = sum(abs(rtts[i] - rtts[i-1]) for i in range(1, len(rtts))) / (len(rtts) - 1) if len(rtts) > 1 else 0
    else:
        avg_rtt = max_rtt = min_rtt = jitter = None
    
    return {
        'host': host,
        'avg_ms': avg_rtt,
        'max_ms': max_rtt,
        'min_ms': min_rtt,
        'jitter_ms': jitter,
        'loss_rate': loss_rate
    }

# 使用示例
stat = ping_stat('8.8.8.8', count=5)
print(stat)

4.4 避坑指南

做 Ping 监控这些年，我踩过不少坑，挑几个典型的说说：

• 防火墙拦截 ICMP：很多云厂商默认禁 Ping。比如 AWS EC2 的安全组，你得手动放行 ICMP。我曾经部署完监控脚本，发现所有云主机都「离线」，结果是被安全组拦了。
• 权限问题：Linux 上普通用户跑 ping3 会报 PermissionError。解决方案：sudo setcap cap_net_raw+ep /usr/bin/python3，或者直接用 root 跑。
• 并发数控制：线程开太多，本地 socket 资源会耗尽。我一般控制在 20 以内。如果目标数量上千，建议用异步 I/O（如 asyncio）替代线程池。
• 时间戳精度：ping3 内部用 time.time()，精度在毫秒级。如果要做微秒级延迟分析，建议改用 time.perf_counter() 自己实现。

4.5 小结