SpringBoot 中使用Spring integration实现Mqtt客户端

最近项目中需要用到mqtt用来发送和接收数据,简单记录以便日后查阅。
!!!文档中有不妥之处请指正。
!!!欢迎共同交流。

1.导入依赖

<!-- MQTT -->
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-core</artifactId>
    <version>5.2.8.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-integration</artifactId>
    <version>2.3.0.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-stream</artifactId>
    <version>5.2.8.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
    <version>5.2.8.RELEASE</version>
</dependency>

2.程序入口

@SpringBootApplication
public class MqttApplication {
    public static void main(String[] args) {
        SpringApplication.run(MqttApplication.class, args);

3.配置类

#application.yaml:
spring:
  mqtt:
    username: username
    password: password
    url: tcp://ip:port
    clientId: mqttclient
    topic: topic1
    completionTimeout: 2000
//configuration
import lombok.Data;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.*;
import org.springframework.integration.mqtt.core.*;
@Data
@Configuration
@ConfigurationProperties(prefix = "spring.mqtt")
public class MqttConfiguration {
    private String username;
    private String password;
    private String url;
        private String cluientId;
    private String topic = "TOPIC_DEFAULT";
    private Integer completionTimeout = 2000;
     * 注册MQTT客户端工厂
     * @return
    @Bean
    public MqttPahoClientFactory mqttClientFactory() throws MqttException {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setConnectionTimeout(0);
        options.setKeepAliveInterval(90);
        options.setAutomaticReconnect(true);
        options.setUserName(this.getUsername());
        options.setPassword(this.getPassword().toCharArray());
        options.setServerURIs(new String[]{this.getUrl()});
        factory.setConnectionOptions(options);
        return factory;

4.配置入站-订阅消息

@Slf4j
@AllArgsConstructor
@Configuration
@IntegrationComponentScan
public class MqttInboundConfiguration {
    private MqttConfiguration mqttConfig;
    private MqttPahoClientFactory factory;
    private MqttMessageReceiver mqttMessageReceiver;
     * 此处可以使用其他消息通道
     * Spring Integration默认的消息通道,它允许将消息发送给一个订阅者,然后阻碍发送直到消息被接收。
     * @return
    @Bean
    public MessageChannel mqttInBoundChannel() {
        return new DirectChannel();
     * 适配器, 两个topic共用一个adapter
     * 客户端作为消费者,订阅主题,消费消息
     * @param
     * @param
     * @return
    @Bean//name = MqttConfiguration.ADAPTER_NAME_IN)
    public MessageProducerSupport mqttInbound() {
        MqttPahoMessageDrivenChannelAdapter
            adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId() + System.currentTimeMillis(),
                                                                factory,
                                                                StringUtils.split(mqttConfig.getConsumer().getTopic(), ","));
        adapter.setCompletionTimeout(60000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setRecoveryInterval(10000);
        adapter.setQos(0);
        adapter.setOutputChannel(mqttInBoundChannel());
        return adapter;
     * mqtt入站消息处理工具,对于指定消息入站通道接收到生产者生产的消息后处理消息的工具。
     * @return
    @Bean
    @ServiceActivator(inputChannel = "mqttInBoundChannel")
    public MessageHandler mqttMessageHandler() {
        return this.mqttMessageReceiver;

消息接收器

@Slf4j @AllArgsConstructor @Component public class MqttMessageReceiver implements MessageHandler { private MqttConfiguration mqttConfig; private MessageProducerSupport adapter; @Override public void handleMessage(Message<?> message) throws MessagingException { try { MessageHeaders headers = message.getHeaders(); //获取消息Topic String receivedTopic = (String) headers.get(MqttHeaders.RECEIVED_TOPIC); log.info("[获取到的消息的topic :]{} ", receivedTopic); //获取消息体 String payload = (String) message.getPayload(); log.info("[获取到的消息的payload :]{} ", payload);*/ } catch (Exception e) { e.printStackTrace(); * 添加主题 * @param topic public void addTopic(String topic){ if (adapter == null){ adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId() + System.currentTimeMillis(), factory, ""); adapter.addTopic(topic,0); log.info("添加 :"+ topic); * 移除主题 * @param * @param payload public void removeTopic(String topic){ if (adapter == null){ adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId() + System.currentTimeMillis(), factory, ""); log.info("移除 :"+topic); adapter.removeTopic(topic);

5.配置出站-发布消息

@Slf4j @AllArgsConstructor @Configuration public class MqttOutboundConfiguration { private MqttConfiguration mqttConfiguration; private MqttPahoClientFactory factory; @Bean public IntegrationFlow mqttOutFlow() { return IntegrationFlows.from(CharacterStreamReadingMessageSource.stdin(), e -> e.poller(Pollers.fixedDelay(2000))) .transform(p -> p + "") .handle(mqttOutbound()) .get(); @Bean @ServiceActivator(inputChannel = "mqttOutboundChannel") public MessageHandler mqttOutbound() { MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfiguration.getProducer().getClientId() + System.currentTimeMillis(), factory); //开启异步 messageHandler.setDefaultQos(0); messageHandler.setAsync(true); messageHandler.setDefaultTopic(mqttConfiguration.getDefaultTopic()); return messageHandler; @Bean//(name = MqttConfiguration.CHANNEL_NAME_OUT) public MessageChannel mqttOutboundChannel() { return new DirectChannel();

消息发送器

* @Description 消息网关用于完成消息的发送 * @Author dongfenglong * @CreateTime 2020-11-10 @Component @MessagingGateway(defaultRequestChannel = MqttConfiguration.CHANNEL_NAME_OUT) public interface MqttGateway { * @param topic * @param payload void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload); * @param topic * @param qos 对消息处理的几种机制。 * * 0 表示的是订阅者没收到消息不会再次发送,消息会丢失。<br> * * 1 表示的是会尝试重试,一直到接收到消息,但这种情况可能导致订阅者收到多次重复消息。<br> * * 2 多了一次去重的动作,确保订阅者收到的消息有一次。 * @param payload json串 void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload); void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, byte[] payload);
@AllArgsConstructor
@Component
public class MqttMessageSender {
    private MqttGateway mqttGateway;
    public void send(String topic, String message) {
        mqttGateway.sendToMqtt(topic, message);
     * @param topic
     * @param messageBody
    public void send(String topic, int qos, MqttMessageBody messageBody){
        mqttGateway.sendToMqtt(topic, qos, messageBody.toString());