MQTT協議學習與在Java（Android通用）中的使用

寫在前面

最近有需求要了解一下各個推送的協議，目前瞭解到實現推送的三個主要方式：MQTT、XMPP和Google Cloud Message（GCM）。第三種方式暫不研究，前兩種都要看一看，本篇討論一下MQTT協議吧。本文使用阿里雲Ubuntu雲服務器安裝代理服務器，使用eclipse paho實現的MqttClient編寫代碼。文中的所使用的帳戶名和密碼在本文發佈後將會更改，請各位自行搭建環境。本文包括如下內容：

• MQTT簡介
• MQTT優點
• MQTT開發環境搭建
• 使用PAHO實現MQTT推送

MQTT簡介 & MQTT優點

MQTT全稱是Message Queuing Telemetry Transport，MQTT是IBM開發的基於TCP/IP協議的輕量級通信協議。MQTT是一個客戶端服務端架構的發佈-訂閱（publish-subscribe）的消息傳輸協議。它的設計思想是開放、簡單、輕量、易於實現。這些特色使它適用於受限環境。例如，但不只限於：

• 網絡代價昂貴，帶寬低、不可靠
• 在嵌入式設備中運行，處理器和內存資源有限

做爲一個物聯網專業的畢業生，看了以上的描述已經心動了，很適合做爲傳感器節點之間的通信協議哇！哦，忘了，我如今是個Androider……MQTT控制報文頭部僅有2字節的長度，下降了網絡傳輸所須要的流量。MQTT支持三種不一樣級別的服務質量（Quality of Service，QoS）爲不一樣場景提供消息可靠性：

• 級別0：盡力而爲。消息發送者會想盡辦法發送消息，可是遇到意外並不會重試。
• 級別1：至少一次。消息接受者若是沒有知會或者知會自己丟失，消息發送者會在此發送以保證消息接收者至少會收到一次，固然可能形成重複消息。
• 級別2：剛好一次。保證這種語義確定會減小併發或者增長延時，不過丟失或者重複消息是不可接受的時候，級別2是最合適的。

若是各位讀完了這些仍然以爲不過癮，沒有戳中各位的痛點，能夠去讀一下MQTT的協議規範，這裏中英文版本都有挑本身愛看的讀一下就好。

MQTT開發環境搭建

首先須要一個代理服務器，這裏mqtt代理服務器使用的是apache的apollo，apollo支持STOMP，AMQP，MQTT，Openwire，SSL和WebSockets。下載戳這。

下載到本地以後，將之上傳到服務器上：

$ scp 文件名 $username@$ip:~
複製代碼

解壓tar.gz：

$ tar zxvf apache-apollo-1.7.1-unix-distro.tar.gz
複製代碼

進入解壓後的bin目錄下執行apollo create testbroker命令建立一個名稱爲testbroker的代理服務器。

$ cd apache-apollo-1.7.1-unix-distro.tar.gz/bin/
$ ./apollo create testbroker
複製代碼

輸入ls命令就能夠看到文件夾下多了一個testbroker的文件夾

進入testbroker的bin文件夾下，執行apollo-broker run 啓動代理服務器。進入testbroker文件下的etc文件夾，能夠看到名爲users.properties的文件，能夠看到在最後配置了用戶名和密碼：

該文件夾下的apollo.xml中配置了端口和ip，不過這裏就無論了。代理服務器配置完畢，接下來就是下載paho實現的mqtt client的jar包了。 下載地址

使用PAHO實現MQTT推送

這裏利用Idea編寫Java程序實現，對於Android程序來講只須要稍加修改就可直接使用。首先新建一個Java項目，接着將上面下載的jar包做爲依賴導入。首先編寫服務端：

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

public class MqttServer {
    /** * 代理服務器ip地址 */
    public static final String MQTT_BROKER_HOST = "tcp://xiasuhuei321.com:61613";

    /** * 訂閱標識 */
    public static final String MQTT_TOPIC = "test";

    private static String userName = "admin";
    private static String password = "password";

    /** * 客戶端惟一標識 */
    public static final String MQTT_CLIENT_ID = "android_server_xiasuhuei321";
    private static MqttTopic topic;
    private static MqttClient client;

    public static void main(String... args) {
        // 推送消息
        MqttMessage message = new MqttMessage();
        try {
            client = new MqttClient(MQTT_BROKER_HOST, MQTT_CLIENT_ID, new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setUserName(userName);
            options.setPassword(password.toCharArray());
            options.setConnectionTimeout(10);
            options.setKeepAliveInterval(20);

            topic = client.getTopic(MQTT_TOPIC);

            message.setQos(1);
            message.setRetained(false);
            message.setPayload("message from server".getBytes());
            client.connect(options);

            while (true) {
                MqttDeliveryToken token = topic.publish(message);
                token.waitForCompletion();
                System.out.println("已經發送");
                Thread.sleep(10000);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

複製代碼

這裏的邏輯很是簡單，建立一個MqttClient，每十秒發送一次消息，訂閱了相應topic的客戶端將會收到這個消息。接下來編寫客戶端：

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

public class MyMqttClient {
    /** * 代理服務器ip地址 */
    public static final String MQTT_BROKER_HOST = "tcp://xiasuhuei321.com:61613";

    /** * 客戶端惟一標識 */
    public static final String MQTT_CLIENT_ID = "android_xiasuhuei321";

    /** * 訂閱標識 */
    public static final String MQTT_TOPIC = "xiasuhuei321";

    /** * */
    public static final String USERNAME = "admin";
    /** * 密碼 */
    public static final String PASSWORD = "password";

    private volatile static MqttClient mqttClient;
    private static MqttConnectOptions options;

    public static void main(String... args) {
        try {
            // host爲主機名，clientid即鏈接MQTT的客戶端ID，通常以客戶端惟一標識符表示，
            // MemoryPersistence設置clientid的保存形式，默認爲之內存保存
            // 設備id不要太騷氣！！！！！！！
            mqttClient = new MqttClient(MQTT_BROKER_HOST, MQTT_CLIENT_ID, new MemoryPersistence());
            // 配置參數信息
            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);
            // 鏈接
            mqttClient.connect(options);
            // 訂閱
            mqttClient.subscribe("test");
            // 設置回調
            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable throwable) {
                    System.out.println("connectionLost");
                }

                @Override
                public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                    System.out.println("Topic: " + s + " Message: " + mqttMessage.toString());
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
複製代碼

接下來啓動服務端和客戶端看一下效果

思考

到這也差很少了，說實話，在Android中難的歷來都不是實現推送，而是如何保證接收推送的服務存活。在Android對後臺服務限制愈來愈大的如今，本身實現推送的意義可能並非很是大。可是對於一些特殊的應用場景下，好比用戶打開應用進行的一些操做須要用到長鏈接，本身實現推送可能會更加可靠一些（聽朋友說三方推送有時會莫名其妙收不到推送）。