Spring?boot怎么集成MQTT

這篇“Spring boot怎么集成MQTT”文章的知識(shí)點(diǎn)大部分人都不太理解，所以小編給大家總結(jié)了以下內(nèi)容，內(nèi)容詳細(xì)，步驟清晰，具有一定的借鑒價(jià)值，希望大家閱讀完這篇文章能有所收獲，下面我們一起來(lái)看看這篇“Spring boot怎么集成MQTT”文章吧。

一、簡(jiǎn)介

MQTT(Message Queuing Telemetry Transport，消息隊(duì)列遙測(cè)傳輸協(xié)議)，是一種基于發(fā)布/訂閱(publish/subscribe)模式的"輕量級(jí)"通訊協(xié)議，可以以極少的代碼和有限的帶寬為連接遠(yuǎn)程設(shè)備提供實(shí)時(shí)可靠的消息服務(wù)。目前在物聯(lián)網(wǎng)、小型設(shè)備、移動(dòng)應(yīng)用等方面有較廣泛的應(yīng)用。

二、主要特性

• （1）使用發(fā)布/訂閱消息模式，提供一對(duì)多的消息發(fā)布，解除應(yīng)用程序耦合。

• （2）對(duì)負(fù)載內(nèi)容屏蔽的消息傳輸。

• （3）使用TCP/IP提供網(wǎng)絡(luò)連接。

• （4）有三種消息發(fā)布服務(wù)質(zhì)量：

“至多一次”，消息發(fā)布完全依賴底層TCP/IP網(wǎng)絡(luò)。會(huì)發(fā)生消息丟失或重復(fù)。這一級(jí)別可用于如下情況，環(huán)境傳感器數(shù)據(jù)，丟失一次讀記錄無(wú)所謂，因?yàn)椴痪煤筮€會(huì)有第二次發(fā)送。這一種方式主要普通APP的推送，倘若你的智能設(shè)備在消息推送時(shí)未聯(lián)網(wǎng)，推送過(guò)去沒(méi)收到，再次聯(lián)網(wǎng)也就收不到了。

“至少一次”，確保消息到達(dá)，但消息重復(fù)可能會(huì)發(fā)生。

“只有一次”，確保消息到達(dá)一次。在一些要求比較嚴(yán)格的計(jì)費(fèi)系統(tǒng)中，可以使用此級(jí)別。在計(jì)費(fèi)系統(tǒng)中，消息重復(fù)或丟失會(huì)導(dǎo)致不正確的結(jié)果。這種最高質(zhì)量的消息發(fā)布服務(wù)還可以用于即時(shí)通訊類的APP的推送，確保用戶收到且只會(huì)收到一次。

• （5）小型傳輸，開(kāi)銷很?。ü潭ㄩL(zhǎng)度的頭部是2字節(jié)），協(xié)議交換最小化，以降低網(wǎng)絡(luò)流量。

• （6）使用Last Will和Testament特性通知有關(guān)各方客戶端異常中斷的機(jī)制。

Last Will：即遺言機(jī)制，用于通知同一主題下的其他設(shè)備發(fā)送遺言的設(shè)備已經(jīng)斷開(kāi)了連接。

Testament：遺囑機(jī)制，功能類似于Last Will。

三、集成步驟

1.引入相關(guān)jar包

  <dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency>
   <groupId>org.springframework.integration</groupId>
   <artifactId>spring-integration-mqtt</artifactId>
</dependency>

2.核心配置類

@Configuration
public class MqttConfig
{
    @Autowired
    private MqttProperties mqttProperties;
    /**
     * 連接器
     * @return
     */
    @Bean
    public MqttConnectOptions getMqttConnectOptions()
    {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        // 設(shè)置是否清空session，false表示服務(wù)器會(huì)保留客戶端的連接記錄，true表示每次連接到服務(wù)器都以新的身份連接
        mqttConnectOptions.setCleanSession(true);
        // 設(shè)置超時(shí)時(shí)間，默認(rèn)30秒
        mqttConnectOptions.setConnectionTimeout(mqttProperties.getConnectionTimeOut());
        mqttConnectOptions.setKeepAliveInterval(mqttProperties.getKeepAlive());
        mqttConnectOptions.setAutomaticReconnect(true);
        // 設(shè)置連接的用戶名
        mqttConnectOptions.setUserName(mqttProperties.getUsername());
        // 設(shè)置連接的密碼
        mqttConnectOptions.setPassword(mqttProperties.getPassword().toCharArray());
        //服務(wù)器地址
        mqttConnectOptions.setServerURIs(new String[]{mqttProperties.getUrl()});
        mqttConnectOptions.setKeepAliveInterval(2);
        return mqttConnectOptions;
    }
    /***
     * MQTT客戶端
     * @return
     */
    @Bean("mqttClientFactory")
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectOptions());
        return factory;
    }
   
    /*-----------------     消息生產(chǎn)者的配置       ---------------------*/
    /**
     * MQTT生產(chǎn)端發(fā)布處理器
     *
     * @return {@link org.springframework.messaging.MessageHandler}
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttProperties.getProducerClientId(), mqttClientFactory());
        messageHandler.setAsync(true);
        return messageHandler;
    }
    /**
     *  MQTT生產(chǎn)端發(fā)布通道
     * @return
     */
    @Bean("mqttOutboundChannel")
    public MessageChannel mqttOutboundChannel() 
    {
        return new DirectChannel();
    }
    /*-----------------   消息消費(fèi)者的配置       ---------------------*/
    
    /**
     * MQTT消費(fèi)端訂閱通道
     *
     * @return {@link org.springframework.messaging.MessageChannel}
     */
    @Bean(name = "mqttInboundChannel")
    public MessageChannel mqttInboundChannel() {
      return new DirectChannel();
    }

    /**
     * MQTT消費(fèi)端連接配置
     *
     * @param channel {@link org.springframework.messaging.MessageChannel}
     * @param factory {@link org.springframework.integration.mqtt.core.MqttPahoClientFactory}
     * @return {@link org.springframework.integration.core.MessageProducer}
     */
    @Bean
    public MessageProducer inbound(
        @Qualifier("mqttInboundChannel") MessageChannel channel,
        @Qualifier("mqttClientFactory") MqttPahoClientFactory factory) {
      MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getConsumerClientId(), factory, "test");
      adapter.setCompletionTimeout(30000);
      adapter.setConverter(new DefaultPahoMessageConverter());
      // 0 至多一次，數(shù)據(jù)可能丟失
      // 1 至少一次，數(shù)據(jù)可能重復(fù)
      // 2 只有一次，且僅有一次，最耗性能
      adapter.setQos(1);
      // 設(shè)置訂閱通道
      adapter.setOutputChannel(channel);
      return adapter;
    }
}

@ConfigurationProperties("mqtt")
@Component
public class MqttProperties implements Serializable
{
    private static final long serialVersionUID = -1425980007744001158L;
    private String url;
    private String username;
    private String password;
    private int keepAlive;
    private int connectionTimeOut;
    private String producerClientId;
    private String producerQos;
    private String consumerClientId;
    private String consumerQos;
    private String consumerTopic;
    private int completionTimeout;
    private String defaultTopic;
    //get、set方法省略
  }

3.網(wǎng)關(guān)配置

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway
{
   void sendToMqtt(byte[] data,@Header(MqttHeaders.TOPIC) String topic);
}

4.編寫測(cè)試類

  @Autowired
  private MqttGateway mqttGateway;
  @RequestMapping("/sendTest")
  public String sendMqttTest(String msg)
  {
      mqttGateway.send("test",msg);
      return "OK";
  }

5.yml配置信息

mqtt:
  url: tcp://localhost:1883
  username: test
  password: test1234
  keep-alive: 30
  connection-timeout: 3000
  producerClientId:  test-producer
  producerQos: 1
  consumerClientId: test-consumer
  consumerQos: 1
  deafultTopic : test

以上就是關(guān)于“Spring boot怎么集成MQTT”這篇文章的內(nèi)容，相信大家都有了一定的了解，希望小編分享的內(nèi)容對(duì)大家有幫助，若想了解更多相關(guān)的知識(shí)內(nèi)容，請(qǐng)關(guān)注億速云行業(yè)資訊頻道。