Android MQTT的訂閱和發布消息

MQTT協議簡述

MQTT(Message Queuing Telemetry Transport，消息隊列遙測傳輸協議)，是基於發布/訂閱(Publish/Subscribe)模式的"輕量級"通訊協議,該協議構建於TCP/IP協議上，有IBM在1999年發布.MQTT最大的優點在於：可以以極少的代碼和有限的寬帶,為連接遠程設備提供可靠的消息服務.。作為一種開銷、低寬帶占用的即時通訊協議，在使其物聯網、小型設備、移動應用等方面較廣泛的應用。

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

 

特點

 

1、實現簡單

2、提供數據傳輸的Qos

3、輕量、占用寬帶低

4、有三種消息發布服務質量  

• "至多一次"，消息發布完成依賴底層TCP/IP網絡。會發生消息丟失或重復。這一級可用於如下情況，壞境傳感器數據，丟失一次讀記錄無所謂，因為不久后還會有第二次發送
• "至少一次"，確保消息到達，但消息重復可能會發生。
• "只有一次",確保消息到達一次，這一級別可用於如下情況，在計費系統中，消息重復或丟失會導致不正確的結果。

5、可傳輸任意類型的數據

6、可保持的會話（session）

 

MQTT協議原理

1 MQTT協議實現方式

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

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

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

2 網絡傳輸與應用消息

MQTT會構建底層網絡傳輸：它將建立客戶端到服務器的連接，提供兩者之間的一個有序的、無損的、基於字節流的雙向傳輸。

當應用數據通過MQTT網絡發送時，MQTT會把與之相關的服務質量（QoS）和主題名（Topic）相關連。

3 MQTT客戶端

一個使用MQTT協議的應用程序或者設備，它總是建立到服務器的網絡連接。客戶端可以：

• （1）發布其他客戶端可能會訂閱的信息；
• （2）訂閱其它客戶端發布的消息；
• （3）退訂或刪除應用程序的消息；
• （4）斷開與服務器連接。

4 MQTT服務器

MQTT服務器以稱為"消息代理"（Broker），可以是一個應用程序或一台設備。它是位於消息發布者和訂閱者之間，它可以：

• （1）接受來自客戶的網絡連接；
• （2）接受客戶發布的應用信息；
• （3）處理來自客戶端的訂閱和退訂請求；
• （4）向訂閱的客戶轉發應用程序消息。

5 MQTT協議中的訂閱、主題、會話

一、訂閱（Subscription）

訂閱包含主題篩選器（Topic Filter）和最大服務質量（QoS）。訂閱會與一個會話（Session）關聯。一個會話可以包含多個訂閱。每一個會話中的每個訂閱都有一個不同的主題篩選器。

二、會話（Session）

每個客戶端與服務器建立連接后就是一個會話，客戶端和服務器之間有狀態交互。會話存在於一個網絡之間，也可能在客戶端和服務器之間跨越多個連續的網絡連接。

三、主題名（Topic Name）

連接到一個應用程序消息的標簽，該標簽與服務器的訂閱相匹配。服務器會將消息發送給訂閱所匹配標簽的每個客戶端。

四、主題篩選器（Topic Filter）

一個對主題名通配符篩選器，在訂閱表達式中使用，表示訂閱所匹配到的多個主題。

五、負載（Payload）

消息訂閱者所具體接收的內容。

6 MQTT協議中的方法

MQTT協議中定義了一些方法（也被稱為動作），來於表示對確定資源所進行操作。這個資源可以代表預先存在的數據或動態生成數據，這取決於服務器的實現。通常來說，資源指服務器上的文件或輸出。主要方法有：

• （1）Connect。等待與服務器建立連接。
• （2）Disconnect。等待MQTT客戶端完成所做的工作，並與服務器斷開TCP/IP會話。
• （3）Subscribe。等待完成訂閱。
• （4）UnSubscribe。等待服務器取消客戶端的一個或多個topics訂閱。
• （5）Publish。MQTT客戶端發送消息請求，發送完成后返回應用程序線程。

 

 

Android 下如何使用MQTT協議

• 導入mqtt包
• 配置MqttConnectOptions
• 調用connect並將配置好的參數寫入
• 通過指定的消息進行消息訂閱
• 向訂閱topic對中發布消息
• 通過mqttCallBack的回調對接收到的消息進行處理

導入mqtt包

implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0' 
implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'

代碼

public class MQTTService extends Service {

    public static final String TAG = MQTTService.class.getSimpleName();

    private static MqttAndroidClient client;
    private MqttConnectOptions conOpt;

    private String host = "tcp://xxx";   //服務器地址 
    private String userName = "xxx";   //賬號
    private String passWord = " xxx";   //密碼
    private static String myTopic = "topic";   //頻道名
    private String clientId = "mqtt_client";   //客戶端ID

    @Override
    public int onStartCommand(Intent intent, int flags, int startId) {
        init();
        return super.onStartCommand(intent, flags, startId);
    }
　　//發布消息 msg
    public static void publish(String msg){
        String topic = myTopic;
        Integer qos = 2;
        Boolean retained = false;
        try {
            client.publish(topic, msg.getBytes(), qos.intValue(), retained.booleanValue());
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    private void init() {
        // 服務器地址（協議+地址+端口號）
        String uri = host;
        client = new MqttAndroidClient(this, uri, clientId);
        // 設置MQTT監聽並且接受消息
        client.setCallback(mqttCallback);

        conOpt = new MqttConnectOptions();
        // 清除緩存
        conOpt.setCleanSession(true);
        // 設置超時時間，單位：秒
        conOpt.setConnectionTimeout(10);
        // 心跳包發送間隔，單位：秒
        conOpt.setKeepAliveInterval(20);
        // 用戶名
        conOpt.setUserName(userName);
        // 密碼
        conOpt.setPassword(passWord.toCharArray());

        // last will message
        boolean doConnect = true;
        String message = "{\"terminal_uid\":\"" + clientId + "\"}";
        String topic = myTopic;
        Integer qos = 2;
        Boolean retained = false;
        if ((!message.equals("")) || (!topic.equals(""))) {
            // 最后
            try {
                conOpt.setWill(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
            } catch (Exception e) {
                Log.i(TAG, "Exception Occured", e);
                doConnect = false;
                iMqttActionListener.onFailure(null, e);
            }
        }

        if (doConnect) {
            doClientConnection();
        }

    }

    @Override
    public void onDestroy() {
        try {
            client.disconnect();  //服務銷毀,斷開連接
        } catch (MqttException e) {
            e.printStackTrace();
        }
        super.onDestroy();
    }

    /** 連接MQTT服務器 */
    private void doClientConnection() {
        if (!client.isConnected() && isConnectIsNomarl()) {
            try {
                client.connect(conOpt, null, iMqttActionListener);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }

    }

    // MQTT是否連接成功
    private IMqttActionListener iMqttActionListener = new IMqttActionListener() {

        @Override
        public void onSuccess(IMqttToken arg0) {
            Log.i(TAG, "連接成功 ");
            try {
                // 訂閱myTopic話題,當訂閱多條頻道,需要遍歷逐條訂閱,否則有可能訂閱失敗
                client.subscribe(myTopic,1);  
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void onFailure(IMqttToken arg0, Throwable arg1) {
            arg1.printStackTrace();
            // 連接失敗，重連
        }
    };

    // MQTT監聽並且接受消息
    private MqttCallback mqttCallback = new MqttCallback() {

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {

            String str1 = new String(message.getPayload());
            MQTTMessage msg = new MQTTMessage(); //自定義接口
            msg.setMessage(str1);
        　　//訂閱信息,接收的信息message
            String str2 = topic + ";qos:" + message.getQos() + ";retained:" + message.isRetained();
            Log.i(TAG, "messageArrived:" + str1);
            Log.i(TAG, str2);
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken arg0) {

        }

        @Override
        public void connectionLost(Throwable arg0) {
            // 失去連接，重連
        }
    };

    /** 判斷網絡是否連接 */
    private boolean isConnectIsNomarl() {
        ConnectivityManager connectivityManager = (ConnectivityManager) this.getApplicationContext().getSystemService(Context.CONNECTIVITY_SERVICE);
        NetworkInfo info = connectivityManager.getActiveNetworkInfo();
        if (info != null && info.isAvailable()) {
            String name = info.getTypeName();
            Log.i(TAG, "MQTT當前網絡名稱：" + name);
            return true;
        } else {
            Log.i(TAG, "MQTT 沒有可用網絡");
            return false;
        }
    }

    @Nullable
    @Override
    public IBinder onBind(Intent intent) {
        return null;
    }
}

 

　連接成功后，就可以實現和服務端消息的發送和接收。

最后在AndroidManifest.xml文件

注冊Services

<service android:name="org.eclipse.paho.android.service.MqttService" />   <!--導入包,Android自帶Mqtt的服務,需要注冊-->
<service android:name=".mqtttest.MQService"/>         <!--自己開啟的服務-->

權限

    <uses-permission android:name="android.permission.WAKE_LOCK" />
    <uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE" />
    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
    <uses-permission android:name="android.permission.READ_PHONE_STATE" />
    <uses-permission android:name="android.permission.READ_EXTERNAL_STORAGE" />
    <uses-permission android:name="android.permission.INTERNET" />

 

完結!

 

最后:請留下您的贊!阿里嘎多.