MQTT 2——服務端安裝與客戶端測試

本篇記錄一下MQTT服務器，客戶端，JAVA客戶端的選擇開發與測試

MQTT服務端選擇與安裝過程；
MQTT客戶端測試工具安裝與測試；
MQTT JAVA客戶端的選擇與開發，測試

---

MQTT服務器選擇與安裝

目前主流的開源MQTT服務器主要是以下3個：
1) EMQX:github 4882 stars
2) Mosquitto：github 1645 stars
3) Apollo：ActiveMQ的升級版，github 109 stars

我們選擇EMQX作為本次MQTT的服務器，首先去EMQ官網去下載EMQX安裝版,截止目前最新版本已到V4.0.4

安裝步驟請參考官網給出的幫助文檔

按照官網操作無法啟動服務，卡在emqx start看過來
  這步很坑的，一定要需要系統先安裝好ERLANG開發環境，ERLANG跳轉地址,可能是因為EMQX是用ERLANG來解釋或開發的

安裝啟動完成后，如果是安裝在本機，可以在瀏覽器中打開http://127.0.0.1:18083，輸入默認用戶名“admin”和默認密碼“public”，能順利進入管理控制台說明EMQX服務器安裝完成。

MQTT客戶端安裝與測試

客戶端測試工具建議使用MQTTBox

MQTTBox有兩種使用的方式：一種為Chrome插件；另外一種為Windows程序安裝。

1) Chrome插件直接去Chrome插件市場搜索mqttbox即可(需要FQ)
2) Windows安裝包位置：點擊下載

連接MQTT服務器：

參考下圖設置連接服務器，協議選擇mqtt/tcp ; Host端口為1883；
其它參數暫不影響測試，有興趣自行研究。

成功連接服務器，Connected圖標會顯示為綠色

增加一個訂閱者

訂單topic為 "hello"；Qos 為默認0

增加一個發布者

發布一個topic為"hello" 的一段消息{"hello":"world"}，Qos默認為0

消息發布后，訂閱者窗口馬上會收到這段消息。至此，客戶端測試就完成。

MQTT JAVA客戶端開發與測試

JAVA客戶端選擇
EMQX SDK_TOOLS上羅列的幾乎所有開發語言下的開源客戶端工具。JAVA下的SDK目前主要有Eclipse Paho,Xenqtt,MeQanTT,mqtt-client。

因為Spring 的 Integration Endpoints去連接官方支持DEMO里面用的就是Eclipse Paho,所以本次JAVA客戶端我們選擇的也是Eclipse Paho .

POM引入

  <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.0</version>
        </dependency>

 

發布端開發示例

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * 發布端未例
 */
public class PublishSample {
    public static void main(String[] args) {

        String topic = "mqtt/okok";
        String content = "hello 哈哈1";
        int qos = 1;
        String broker = "tcp://127.0.0.1:1883";
        String userName = "test";
        String password = "test";
        String clientId = "pubClient1";
        // 內存存儲
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            // 創建客戶端
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            // 創建鏈接參數
            MqttConnectOptions connOpts = new MqttConnectOptions();
            // 在重新啟動和重新連接時記住狀態
            connOpts.setCleanSession(false);
            // 設置連接的用戶名
            connOpts.setUserName(userName);
            connOpts.setPassword(password.toCharArray());
            // 建立連接
            sampleClient.connect(connOpts);
            // 創建消息
            MqttMessage message = new MqttMessage(content.getBytes());
            // 設置消息的服務質量
            message.setQos(qos);
            // 發布消息
            sampleClient.publish(topic, message);
            // 斷開連接
            sampleClient.disconnect();
            // 關閉客戶端
            sampleClient.close();
        } catch (MqttException me) {
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("excep " + me);
            me.printStackTrace();
        }
    }
}

 

訂閱端發布實例

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * 訂閱端
 */
public class SubscribeSample {
    public static void main(String[] args) throws MqttException {
        String HOST = "tcp://127.0.0.1:1883";
        String TOPIC = "mqtt/okok";
        int qos = 1;
        String clientid = "subClient1";
        String userName = "test";
        String passWord = "test";
        try {
            // host為主機名，test為clientid即連接MQTT的客戶端ID，一般以客戶端唯一標識符表示，MemoryPersistence設置clientid的保存形式，默認為以內存保存
            MqttClient client = new MqttClient(HOST, clientid, new MemoryPersistence());
            // MQTT的連接設置
            MqttConnectOptions options = new MqttConnectOptions();
            // 設置是否清空session,這里如果設置為false表示服務器會保留客戶端的連接記錄，這里設置為true表示每次連接到服務器都以新的身份連接
            options.setCleanSession(true);
            // 設置連接的用戶名
            options.setUserName(userName);
            // 設置連接的密碼
            options.setPassword(passWord.toCharArray());
            // 設置超時時間 單位為秒
            options.setConnectionTimeout(10);
            // 設置會話心跳時間 單位為秒 服務器會每隔1.5*20秒的時間向客戶端發送個消息判斷客戶端是否在線，但這個方法並沒有重連的機制
            options.setKeepAliveInterval(20);
            // 設置回調函數
            client.setCallback(new MqttCallback() {

                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("connectionLost");
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("topic:"+topic);
                    System.out.println("Qos:"+message.getQos());
                    System.out.println("message content:"+new String(message.getPayload()));

                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("deliveryComplete---------"+ token.isComplete());
                }

            });
            client.connect(options);
            //訂閱消息
            client.subscribe(TOPIC, qos);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

 

下一篇：MQTT整合進Spring Mvc

更多參考網站：

1.EMQX幫助中心 https://docs.emqx.io/broker/latest/cn/
2.Spring MQTT Support https://docs.spring.io/spring-integration/docs/5.0.8.RELEASE/reference/html/mqtt.html