Mosquitto转发MQTT Protobuff消息出现payload损坏问题如何解决

首先纠正一个误解：Mosquitto 作为标准MQTT broker，不会对任何Payload内容做解码操作，所有MQTT broker默认都是二进制透传Payload，你遇到的Payload损坏完全是Java Spring侧的消息转换逻辑导致的，RabbitMQ正常只是因为其MQTT插件的默认转换配置没有破坏二进制内容。

你当前代码的两个核心错误 #

1. 你调用ByteArrayMessageConverter.fromMessage(message,String.class)的操作，会把二进制protobuf内容用默认字符集（通常为UTF-8）强制解码，不符合UTF-8编码规则的字节会被替换为乱码标识�，这一步直接破坏了原始二进制Payload
2. 你调用parseFrom时强转Payload为ByteBuffer，如果Spring默认消息转换器已经提前把Payload转成了String类型，强转只会拿到错误内容，直接导致解析失败

修复步骤 #

• 第一步：配置Spring MQTT入站通道适配器，设置payload-as-bytes属性为true，或者手动指定DefaultPahoMessageConverter的payloadAsBytes参数为true，禁止Spring提前将二进制Payload转换为字符串
• 第二步：删除代码中无用的转String逻辑，不要对protobuf二进制Payload做任何字符集相关的转换操作
• 第三步：直接用原始字节数组解析protobuf即可

修复后的代码示例 #

@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler messageHandler() {
    var converter = new TelemetryConverter();
    return new MessageHandler() {
        @Override
        public void handleMessage(Message<?> message) throws MessagingException {
            // 直接获取原始二进制payload
            byte[] payload = (byte[]) message.getPayload();
            // 仅打印长度校验，不要直接打印二进制内容，转字符串必然出现乱码
            log.info("Received message length: {}", payload.length);
            try {
                MeasurementProto.Measurement measurement = MeasurementProto.Measurement.parseFrom(payload);
                // 后续业务逻辑
            } catch (InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
            process.handle(converter.fromMessage(message));
        }
    };
}

额外校验方法 #

你可以在Python发送端打印Payload的字节长度，Java接收端也打印对应长度，如果两端长度一致，说明Broker透传完全正常，按上述方法修复即可。

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