由於最近公司在開發一款後臺與安卓的更新系統，經過再三研究之後，也是選擇Mqtt這個目前流行的框架。為了能夠讓專案運營起來，最終雖說是選擇ActiveMQ。但在這個過程中，也是發現Apollo作為伺服器也是相當不錯。當然對於後者已經被apace放棄，不過今天還是和大家整理一下SpringBoot+MQTT+apollo實現訂閱釋出功能的全過程。

對於專案首先需要用到的前提東西，比如Apollo如何下載，以及MQTT測試工具在這裡就不多說。如果真的不懂私聊Damon吧，在這裡就不浪費時間。

對於專案，首先你所需要引入maven包：

pom.xml

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

其目標就是將MQTT用在專案組中
接著就是專案yml檔案的配置，使用properties，以葫蘆畫瓢就行了：

applicaiton.yml

mqtt:
 username: admin
 password: password
 host-url: tcp://127.0.0.1:8161 # 你自己伺服器的地址和埠，這個需要改
 clientID: test1    # 這個改不改隨意，但不同的客戶端肯定不能一樣
 default-topic: home/garden/fountain   # 預設主題
 timeout: 100
 keepalive: 100

# Tomcat
server:
 tomcat:
  uri-encoding: UTF-8
  max-threads: 1000
  min-spare-threads: 30
 port: 8088

注意host-url，這就是你apollo的地址

來到第三步，此時就是專案內的檔案：

MqttConfig檔案

@Component
@ConfigurationProperties("mqtt")
@Setter
@Getter
public class MqttConfig {
  @Autowired
  private MqttPushClient mqttPushClient;

  /**
   * 使用者名稱
   */
  // @Value("username")
  private String username;
  /**
   * 密碼
   */
  private String password;
  /**
   * 連線地址
   */
  private String hostUrl;
  /**
   * 客戶Id
   */
  private String clientID;
  /**
   * 預設連線話題
   */
  private String defaultTopic;
  /**
   * 超時時間
   */
  private int timeout;
  /**
   * 保持連線數
   */
  private int keepalive;

  @Bean
  public MqttPushClient getMqttPushClient() {
    System.out.println("hostUrl: "+ hostUrl);
    System.out.println("clientID: "+ clientID);
    System.out.println("username: "+ username);
    System.out.println("password: "+ password);
    System.out.println("timeout: "+timeout);
    System.out.println("keepalive: "+ keepalive);
    mqttPushClient.connect(hostUrl,clientID,username,password,timeout,keepalive);
    // 以/#結尾表示訂閱所有以test開頭的主題
    mqttPushClient.subscribe(defaultTopic,0);
    return mqttPushClient;
  }
}

目的就是配置所對應的訊息

第四步就是釋出以及訂閱等功能：

MqttPushClient

@Component
public class MqttPushClient {
  private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class);

  @Autowired
  private PushCallback pushCallback;

  private static MqttClient client;

  private static MqttClient getClient() {
    return client;
  }

  private static void setClient(MqttClient client) {
    MqttPushClient.client = client;
  }

  /**
   * 客戶端連線
   *
   * @param host   ip+埠
   * @param clientID 客戶端Id
   * @param username 使用者名稱
   * @param password 密碼
   * @param timeout  超時時間
   * @param keepalive 保留數
   */
  public void connect(String host,String clientID,String username,String password,int timeout,int keepalive) {
    MqttClient client;
    try {
      client = new MqttClient(host,new MemoryPersistence());
      MqttConnectOptions options = new MqttConnectOptions();
      options.setCleanSession(true);
      options.setUserName(username);
      options.setPassword(password.toCharArray());
      options.setConnectionTimeout(timeout);
      options.setKeepAliveInterval(keepalive);
      MqttPushClient.setClient(client);
      try {
        client.setCallback(pushCallback);
        client.connect(options);
      } catch (Exception e) {
        e.printStackTrace();
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /**
   * 釋出
   *
   * @param qos     連線方式
   * @param retained  是否保留
   * @param topic    主題
   * @param pushMessage 訊息體
   */
  public void publish(int qos,boolean retained,String topic,String pushMessage) {
    MqttMessage message = new MqttMessage();
    message.setQos(qos);
    message.setRetained(retained);
    message.setPayload(pushMessage.getBytes());
    MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic);
    if (null == mTopic) {
      logger.error("topic not exist");
    }
    MqttDeliveryToken token;
    try {
      token = mTopic.publish(message);
      token.waitForCompletion();
    } catch (MqttPersistenceException e) {
      e.printStackTrace();
    } catch (MqttException e) {
      e.printStackTrace();
    }
  }

  /**
   * 訂閱某個主題
   *
   * @param topic 主題
   * @param qos  連線方式
   */
  public void subscribe(String topic,int qos) {
    logger.info("開始訂閱主題" + topic);
    try {
      MqttPushClient.getClient().subscribe(topic,qos);
    } catch (MqttException e) {
      e.printStackTrace();
    }
  }
}

訂閱主題以及釋出的方式等內容更多編寫
最後在搞個測試看看我們的結果是否正確：

TestController

@RestController
@RequestMapping("/")
public class TestController {

  @Autowired
  private MqttPushClient mqttPushClient;

  @GetMapping(value = "/publishTopic")
  public String publishTopic() {
    String topicString = "home/garden/fountain";
    mqttPushClient.publish(0,false,topicString,"測試一下發布訊息");
    return "ok";
  }
  // 傳送自定義訊息內容（使用預設主題）
  @RequestMapping("/publishTopic/{data}")
  public String test1(@PathVariable("data") String data) {
    String topicString = "home/garden/fountain";
    mqttPushClient.publish(0,data);
    return "ok";
  }

  // 傳送自定義訊息內容，且指定主題
  @RequestMapping("/publishTopic/{topic}/{data}")
  public String test2(@PathVariable("topic") String topic,@PathVariable("data") String data) {
    mqttPushClient.publish(0,topic,data);
    return "ok";
  }
}

如此一來就OK！

你可以使用MQTT.fx進行測試。用Postman發出，就能夠檢視最終的結果。在這裡，因為時間的原因就不多說，有啥有趣的問題，咱們可以一同探討。 希望你希望，Damon將會不斷的分享各種有趣的開發小故事給大家娛樂。下一期，或在POI實現匯出匯入或者是ActiveMQ進行選擇。

到此這篇關於SpringBoot+MQTT+apollo實現訂閱釋出功能的示例的文章就介紹到這了,更多相關SpringBoot+MQTT+apollo訂閱釋出內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！