使用 tcpdump 分析 Mosquitto 客户端消息时延

1. 抓包准备

• 在运行 Mosquitto 代理的服务器上执行命令：

  tcpdump -i eth0 -w mqtt.pcap 'tcp port 1883'
  

  • -i eth0：指定监听的网络接口（根据实际调整）
  • -w mqtt.pcap：保存抓包数据到文件
  • tcp port 1883：过滤 MQTT 默认端口流量（TLS 场景用 8883）

2. **消息时延测量原理

消息时延 $T_{\text{delay}}$ 由三部分组成： $$T_{\text{delay}} = T_{\text{pub}} + T_{\text{broker}} + T_{\text{sub}}$$ 其中：

• $T_{\text{pub}}$：发布者到代理的网络传输时间
• $T_{\text{broker}}$：代理处理时间
• $T_{\text{sub}}$：代理到订阅者的网络传输时间

3. **关键数据包识别

在抓包文件中需匹配：

1. 发布者 → 代理：PUBLISH 报文（源 IP = 发布者）
2. 代理 → 订阅者：PUBLISH 报文（目的 IP = 订阅者）
3. 匹配依据：
   • 相同的 Topic 路径
   • 相同的 Payload 长度
   • 相同消息的 Packet Identifier（QoS ≥1 时）

4. **时延计算步骤

tshark -r mqtt.pcap -Y "mqtt.msgtype == 3" -T fields -e frame.time_epoch -e ip.src -e ip.dst -e mqtt.topic

1. 提取 PUBLISH 包时间戳（Unix 时间格式）
2. 找到同一消息的两条记录：
   • 第一条：发布者 IP → 代理 IP
   • 第二条：代理 IP → 订阅者 IP
3. 计算时延： $$T_{\text{delay}} = t_{\text{sub_recv}} - t_{\text{pub_send}}$$

5. **Python 自动化示例

from scapy.all import *
import sys

def calc_delay(pcap_file):
    pkts = rdpcap(pcap_file)
    pub_pkts = {}  # 存储发布者消息: {msg_id: timestamp}
    delays = []    # 存储时延结果
    
    for pkt in pkts:
        if pkt.haslayer('MQTT') and pkt['MQTT'].msgtype == 3:  # PUBLISH
            msg_id = pkt['MQTT'].msgid if pkt['MQTT'].qos > 0 else None
            src = pkt['IP'].src
            dst = pkt['IP'].dst
            
            # 发布者发送记录
            if src == "PUB_IP" and dst == "BROKER_IP":  # 替换实际IP
                pub_pkts[msg_id] = pkt.time
                
            # 订阅者接收记录
            elif src == "BROKER_IP" and dst == "SUB_IP":  # 替换实际IP
                if msg_id in pub_pkts:
                    delay = pkt.time - pub_pkts[msg_id]
                    delays.append(delay * 1000)  # 转毫秒
                    print(f"消息 {msg_id} 时延: {delay:.3f} 秒")
    
    print(f"平均时延: {sum(delays)/len(delays):.2f} ms")

if __name__ == "__main__":
    calc_delay(sys.argv[1])

6. **注意事项

1. 时间同步：确保所有设备使用 NTP 时间同步
2. QoS 选择：
   • QoS 0：需通过 Payload 内容匹配消息
   • QoS 1/2：优先使用 Packet ID 匹配
3. 精度限制：网络抖动可能导致 ±10ms 误差
4. 优化建议：
   • 在消息 Payload 中添加唯一时间戳
   • 使用 Wireshark 的 mqtt.time_delta 过滤直接分析

提示：实际测试时建议发送 100+ 条消息获取统计结果，排除偶发干扰。