SpringBoot+MQTT+apollo實現訂閱發佈功能的示例

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

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

對於項目，首先你所需要引入maven包：

pom.xml
 

    org.springframework.integrationspring-integration-streamorg.springframework.integrationspring-integration-mqtt

其目標就是將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, clientID, 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,false,topicString, data);      return "ok";    }      // 發送自定義消息內容，且指定主題    @RequestMapping("/publishTopic/{topic}/{data}")    public String test2(@PathVariable("topic") String topic, @PathVariable("data") String data) {      mqttPushClient.publish(0,false,topic, data);      return "ok";    }  }

如此一來就OK！
 

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