MQTT

新的項目需求,使用到了硬件鏈接,運用到了mqtt技術,查詢了相關資料,總結了一下

1、MQTT簡介

　　MQTT（Message Queuing Telemetry Transport，消息隊列遙測傳輸協議），是一種基於發佈/訂閱（publish/subscribe）模式的輕量級協議，該協議構建於TCP/IP協議之上，MQTT最大優勢在於，能夠以極少的代碼和有限的帶寬，爲鏈接遠程設備提供實時可靠的消息服務。做爲一種低開銷、低帶寬佔用的即時通信協議，使其在物聯網、小型設備、移動應用等方面有較普遍的應用。

　　MQTT是一個基於客戶端-服務器的消息發佈/訂閱傳輸協議。MQTT協議是輕量、簡單、開放和易於實現的，這些特色使它適用範圍很是普遍。在不少狀況下，包括受限的環境中，如：機器與機器（M2M）通訊和物聯網（IoT）。其在，經過衛星鏈路通訊傳感器、偶爾撥號的醫療設備、智能家居、及一些小型化設備中已普遍使用。　　

2、特性

　　MQTT協議工做在低帶寬、不可靠的網絡的遠程傳感器和控制設備通信而設計的協議，它具備如下主要的幾項特性：

　　（1）使用發佈/訂閱消息模式，提供一對多的消息發佈，解除應用程序耦合。

　　（2）對負載內容屏蔽的消息傳輸。

　　（3）使用TCP/IP提供網絡鏈接。

　　主流的MQTT是基於TCP鏈接進行數據推送的，可是一樣有基於UDP的版本，叫作MQTT-SN。這兩種版本因爲基於不一樣的鏈接方式，優缺點天然也就各有不一樣了。

　　（4）有三種消息發佈服務質量(qss)：

　　「至多一次」，消息發佈徹底依賴底層TCP/IP網絡。會發生消息丟失或重複。這一級別可用於以下狀況，環境傳感器數據，丟失一次讀記錄無所謂，由於不久後還會有第二次發送。這一種方式主要普通APP的推送，假若你的智能設備在消息推送時未聯網，推送過去沒收到，再次聯網也就收不到了。

　　「至少一次」，確保消息到達，但消息重複可能會發生。

　　「只有一次」，確保消息到達一次。在一些要求比較嚴格的計費系統中，可使用此級別。在計費系統中，消息重複或丟失會致使不正確的結果。這種最高質量的消息發佈服務還能夠用於即時通信類的APP的推送，確保用戶收到且只會收到一次。　

　　

 

 

 固然了這個數字越小,帶寬帶站越低

3、原理

 

 

實現MQTT協議須要客戶端和服務器端通信完成，在通信過程當中，MQTT協議中有三種身份：發佈者（Publish）、代理（Broker）（服務器）、訂閱者（Subscribe）。其中，消息的發佈者和訂閱者都是客戶端，消息代理是服務器，消息發佈者能夠同時是訂閱者。

MQTT傳輸的消息分爲：主題（Topic）和負載（payload）兩部分：

• （1）Topic，能夠理解爲消息的類型，訂閱者訂閱（Subscribe）後，就會收到該主題的消息內容（payload）；
• （2）payload，能夠理解爲消息的內容，是指訂閱者具體要使用的內容。

4、實現

1.下載emq

 

 

 2.啓動後訪問

localhost:18083 默認的帳號密碼是admin獲得下圖

 

 

 3.經過node red圖形化工具創建鏈接

 

注意這裏的主題就是topic訂閱主題,必定要一致

 

 

 

 

 

 正常流程操做完成後能夠再emq的圖形頁面看到客戶鏈接

 

 發送消息後,確實能夠收到數據賊表示鏈接成功

 

 基本上到這裏簡單mqtt 鏈接就算實現了

後續不想使用emq,也可使用阿里的 原理其實差很少,只不過有幾個id key須要購買,就看我的狀況而定了

        public class MqttPublishSample {

        public static void main(String[] args) {

            String topic        = "MQTT Examples";
            String content      = "Message from MqttPublishSample";
            int qos             = 2;
            String broker       = "tcp://mqtt.eclipse.org:1883";
            String clientId     = "JavaSample";
            MemoryPersistence persistence = new MemoryPersistence();

            try {
                MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
                MqttConnectOptions connOpts = new MqttConnectOptions();
                connOpts.setCleanSession(true);
                System.out.println("Connecting to broker: "+broker);
                sampleClient.connect(connOpts);
                System.out.println("Connected");
                System.out.println("Publishing message: "+content);
                MqttMessage message = new MqttMessage(content.getBytes());
                message.setQos(qos);
                sampleClient.publish(topic, message);
                System.out.println("Message published");
                sampleClient.disconnect();
                System.out.println("Disconnected");
                System.exit(0);
            } 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();
            }
        }
    }
記錄本周的一個粗心錯誤->redis讀取的服務器地址沒有開啓,換成本地的就ok了