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

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

Part1引入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>

其目標(biāo)就是將MQTT用在項目組中

yml文件的配置

使用properties，以葫蘆畫瓢就行了：

applicaiton.yml
mqtt:
  username: admin
  password: password
  host-url: tcp://127.0.0.1:8161  # 你自己服務(wù)器的地址和端口，這個需要改
  clientID: test1       # 這個改不改隨意，但不同的客戶端肯定不能一樣
  default-topic: home/garden/fountain      # 默認(rèn)主題
  timeout: 100
  keepalive: 100

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

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

Part2項目內(nèi)的文件：

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;
    /**
     * 默認(rèn)連接話題
     */
    private String defaultTopic;
    /**
     * 超時時間
     */
    private int timeout;
    /**
     * 保持連接數(shù)
     */
    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);
        // 以/#結(jié)尾表示訂閱所有以test開頭的主題
        mqttPushClient.subscribe(defaultTopic, 0);
        return mqttPushClient;
    }
}

目的就是配置所對應(yīng)的消息

Part3發(fā)布以及訂閱等功能

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 保留數(shù)
     */
    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();
        }
    }

    /**
     * 發(fā)布
     *
     * @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();
        }
    }
}

訂閱主題以及發(fā)布的方式等內(nèi)容更多編寫

Part4測試

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, "測試一下發(fā)布消息");
        return "ok";
    }
    // 發(fā)送自定義消息內(nèi)容（使用默認(rèn)主題）
    @RequestMapping("/publishTopic/{data}")
    public String test1(@PathVariable("data") String data) {
        String topicString = "home/garden/fountain";
        mqttPushClient.publish(0,false,topicString, data);
        return "ok";
    }

    // 發(fā)送自定義消息內(nèi)容，且指定主題
    @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進(jìn)行測試。用Postman發(fā)出，就能夠查看最終的結(jié)果。在這里，因為時間的原因就不多說，有啥有趣的問題，咱們可以一同探討。