Spring Integration发送MQTT消息时contentType头未生效问题

你的header配置逻辑没有错误，问题出在MqttPahoMessageHandler的默认行为和MQTT协议版本限制上：

• 默认的MqttPahoMessageHandler是适配MQTT 3.1/3.1.1版本的处理器，只会透传MQTT协议明确规定的几个标准头（topic、qos、retain、dup），包括MessageHeaders.CONTENT_TYPE在内的Spring Messaging通用头、自定义头默认都会被丢弃，不会随消息发送到broker
• MQTT 3.x版本协议本身没有设计自定义消息属性的字段，根本不支持在消息体之外携带额外的自定义头，你在RabbitMQ中看到的x-mqtt-dup、x-mqtt-publish-qos是RabbitMQ MQTT插件自己解析协议字段后添加的内部属性，不是你自定义的头。

根据你使用的MQTT协议版本选择对应方案：

方案1：使用MQTT 5.0版本（支持自定义属性） #

MQTT 5.0原生支持内容类型、自定义用户属性字段，你需要替换处理器并配置头映射规则：

1. 将MqttPahoMessageHandler替换为Spring Integration提供的MQTT 5.0专属处理器Mqttv5PahoMessageHandler
2. 自定义头映射逻辑，将contentType写入MQTT 5协议规定的contentType属性字段
   示例配置代码：

@Bean
public IntegrationFlow outgoingMqttMsgFlow() {
    Mqttv5PahoMessageHandler mqttHandler = new Mqttv5PahoMessageHandler(mqttBroker, UUID.randomUUID().toString());
    // 配置头映射，将Spring的contentType头写入MQTT5标准属性
    mqttHandler.setHeaderMapper(new DefaultMqttHeaderMapper() {
        @Override
        public void fromHeaders(MessageHeaders headers, MqttMessageProperties mqttProps) {
            super.fromHeaders(headers, mqttProps);
            String contentType = headers.get(MessageHeaders.CONTENT_TYPE, String.class);
            if (contentType != null) {
                mqttProps.getProperties().add(new MqttProperty(MqttPropertyId.CONTENT_TYPE, contentType));
            }
        }
    });

    return IntegrationFlows.from("outgoingMqttMsgChannel")
            .enrichHeaders(headers -> headers.header(MqttHeaders.TOPIC, mqttTopic))
            .enrichHeaders(headers -> headers.header(MessageHeaders.CONTENT_TYPE, "usp.msg", true))
            .handle(mqttHandler)
            .get();
}

注意：需要同时确认RabbitMQ的MQTT插件已经开启MQTT 5.0支持，否则无法正确识别解析该属性。

方案2：继续使用MQTT 3.x版本 #

MQTT 3.x没有自定义属性的承载位置，无法在消息体外单独携带contentType头，你需要将contentType作为消息体的一部分，和业务数据一起序列化后发送。

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