如何解析curl订阅MQTT通配符主题返回的多响应数据？

问题背景 #

用curl订阅MQTT通配符主题tele/#时，会收到无分隔符的连续响应数据流。通过--trace -可见单条响应遵循MQTT格式：

• 2字节主题长度（大端序：第1字节为高位MSB，第2字节为低位LSB）
• n字节主题名称（n由上述主题长度值决定）
• 剩余字节为主题负载

需要用Perl脚本读取curl输出的数据流，拆分出每条完整消息，但无法直接确定负载长度，需通过协议规则解析边界。

核心解析思路 #

订阅后收到的是MQTT的PUBLISH消息，其结构为固定头 + 可变头 + 负载，解析边界的关键在于：

1. 固定头：首字节为消息类型+标志，后续是剩余长度字段（可变字节编码，最多4字节），剩余长度代表「可变头+负载」的总字节数。
2. 可变头：2字节主题长度（大端序） + n字节主题名（n=主题长度）。
3. 负载长度：剩余长度 - 2（主题长度字节数） - n（主题名长度）。

通过先解析剩余长度，就能确定单条消息的总字节数，从而拆分连续数据流。

Perl实现示例 #

以下脚本直接读取curl输出的二进制流，按MQTT协议拆分每条消息：

#!/usr/bin/perl
use strict;
use warnings;

# 替换为你的MQTT服务器地址、认证参数等
open my $curl_fh, '-|', 'curl -s -N --mqtt "mqtt://your-mqtt-server:1883" --mqtt-subscribe "tele/#"' 
    or die "启动curl失败: $!";
binmode $curl_fh; # 二进制模式读取，避免换行符破坏数据结构

while (1) {
    # 1. 读取固定头首字节（消息类型+标志）
    my $fixed_head;
    my $bytes_read = read($curl_fh, $fixed_head, 1);
    last unless $bytes_read == 1; # 数据流结束
    
    # 2. 解析剩余长度（MQTT可变字节编码）
    my $remaining_length = 0;
    my $multiplier = 1;
    my $byte;
    do {
        read($curl_fh, $byte, 1) or last;
        my $val = ord($byte) & 0x7F;
        $remaining_length += $val * $multiplier;
        $multiplier *= 128;
    } while (ord($byte) & 0x80);
    
    # 3. 读取剩余长度对应的可变头+负载字节
    my $var_header_payload;
    read($curl_fh, $var_header_payload, $remaining_length) or last;
    
    # 4. 解析主题和负载
    my $topic_length = unpack('n', substr($var_header_payload, 0, 2)); # 大端序解析16位整数
    my $topic = substr($var_header_payload, 2, $topic_length);
    my $payload = substr($var_header_payload, 2 + $topic_length);
    
    # 输出解析结果
    print "主题: $topic\n";
    print "负载: $payload\n";
    print "-------------------------\n";
}

close $curl_fh;

关键说明 #

• 二进制模式：必须用binmode处理curl输出句柄，防止系统自动转换换行符，破坏二进制数据结构。
• 剩余长度解析：MQTT剩余长度用可变字节编码，每个字节最高位为续位标志，直到最高位为0的字节结束，这是拆分消息的核心。
• 日志过滤：用curl -s参数关闭日志输出，只保留原始MQTT二进制数据流，避免干扰解析。

内容的提问来源于stack exchange，提问作者Steven Christensen