【MQTT】

1. 实体

public class MqttMessagePayload {private String traceId;private String mdcSn;// Getters, Setters, toString...
}

2. 配置

@ConfigurationProperties(prefix = "sdk.mqtt")
public class MqttProperties {private boolean enabled = false;private String brokerUrl;private String clientId;private String topic;// Getters and Setters
}

• 业务使用

sdk:mqtt:enabled: truebroker-url: tcp://localhost:1883client-id: my-client-idtopic: my-topic

3. 配置类

@Configuration
@ConditionalOnProperty(name = "sdk.mqtt.enabled", havingValue = "true")
@EnableConfigurationProperties(MqttProperties.class)
public class MqttAutoConfiguration {@Beanpublic MqttOutboundChannelAdapter mqttOutbound(MqttProperties properties) {// 省略 connectionOptions 配置，可加用户名密码等MqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(properties.getClientId(), factory);messageHandler.setAsync(true);messageHandler.setDefaultTopic(properties.getTopic());return new MqttOutboundChannelAdapter(messageHandler);}
}

4. service

public class MqttMessageSender {private final MessageChannel mqttOutboundChannel;public MqttMessageSender(MessageChannel mqttOutboundChannel) {this.mqttOutboundChannel = mqttOutboundChannel;}public void sendMessage(MqttMessagePayload payload) {try {GenericMessage<String> message = new GenericMessage<>(new ObjectMapper().writeValueAsString(payload));mqttOutboundChannel.send(message);} catch (Exception e) {throw new MqttSendException("MQTT消息发送失败", e);}}
}

5. SDK集成

sdk.mqtt.enabled: true

@Autowired(required = false)
private MqttMessageSender mqttMessageSender;public void someBusinessLogic() {if (mqttMessageSender != null) {mqttMessageSender.sendMessage(new MqttMessagePayload("trace123", "mdc456"));}
}