訂閱和接收

//    final String serverUri = "tcp://iot.eclipse.org:1883";
    final String serverUri = "tcp://ip:port";
    String clientId = "ExampleAndroidClient";
    final String subscriptionTopic = "subscribe_topic";
    final String publishTopic = "publish_topic";
    final String publishMessage = "Hello World!"
 
;
    private static String userName = "admin";
    private static String passWord = "password";
private void initConnect() {
        //封裝好的MQTTClient供操作，傳送和接手等設定都用它
        mqttAndroidClient = new MqttAndroidClient(getApplicationContext(), serverUri, clientId);
        //設定回撥
        mqttAndroidClient.setCallback(new
 
 MqttCallbackExtended() {

            /**
             * 連線完成回撥
             * @param reconnect true 斷開重連,false 首次連線
             * @param serverURI 伺服器URI
             */
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {

                if (reconnect) {
                    addToHistory("Reconnected to : "
 
 + serverURI);
                    // Because Clean Session is true, we need to re-subscribe
                    subscribeToTopic();
                } else {
                    addToHistory("Connected to: " + serverURI);
                }
            }

            /**
             * @desc 連線斷開回調
             * 可在這裡做一些重連等操作
             */
            @Override
            public void connectionLost(Throwable cause) {
                addToHistory("The Connection was lost.");
            }

            /**
             * 訊息接收，如果在訂閱的時候沒有設定IMqttMessageListener，那麼收到訊息則會在這裡回撥。
             * 如果設定了IMqttMessageListener，則訊息回撥在IMqttMessageListener中
             * @param topic 該訊息來自的訂閱主題
             * @param message 訊息內容
             */
            @Override
            public void messageArrived(String topic, MqttMessage message) {
                addToHistory("Incoming message: " + new String(message.getPayload()));
            }

            /**
             * 交付完成回撥。在publish訊息的時候會收到此回撥.
             * qos:
             * 0 傳送完則回撥
             * 1 或 2 會在對方收到時候回撥
             * @param token
             */
            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                addToHistory("deliveryComplete,token:" + token);
            }
        });

        //mqtt連線引數設定
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        //設定自動重連
        mqttConnectOptions.setAutomaticReconnect(true);
        // 設定是否清空session,這裡如果設定為false表示伺服器會保留客戶端的連線記錄
        // 這裡設定為true表示每次連線到伺服器都以新的身份連線
        mqttConnectOptions.setCleanSession(false);
        //設定連線的使用者名稱
        mqttConnectOptions.setUserName(userName);
        //設定連線的密碼
        mqttConnectOptions.setPassword(passWord.toCharArray());
        // 設定超時時間 單位為秒
        mqttConnectOptions.setConnectionTimeout(10);
        // 設定會話心跳時間 單位為秒 伺服器會每隔1.5*20秒的時間向客戶端傳送個訊息判斷客戶端是否線上，但這個方法並沒有重連的機制
        mqttConnectOptions.setKeepAliveInterval(20);
        try {
            //設定好相關引數後，開始連線
            mqttAndroidClient.connect(mqttConnectOptions, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    Log.i(TAG, "onSuccess,asyncActionToken");
                    /*連線成功之後設定連線斷開的緩衝配置*/
                    DisconnectedBufferOptions disconnectedBufferOptions = new DisconnectedBufferOptions();
                    //開啟
                    disconnectedBufferOptions.setBufferEnabled(true);
                    //離線後最多快取100調
                    disconnectedBufferOptions.setBufferSize(100);
                    //不一直持續留存
                    disconnectedBufferOptions.setPersistBuffer(false);
                    //刪除舊訊息
                    disconnectedBufferOptions.setDeleteOldestMessages(false);
                    mqttAndroidClient.setBufferOpts(disconnectedBufferOptions);
                    //訂閱主題
                    subscribeToTopic();
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    addToHistory("Failed to connect to: " + serverUri);
                }
            });


        } catch (MqttException ex) {
            ex.printStackTrace();
        }
    }


    /**
     * 訂閱主題
     */
    public void subscribeToTopic() {
        try {
            //主題、QOS、context,訂閱監聽，訊息監聽
            mqttAndroidClient.subscribe(subscriptionTopic, 0, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    addToHistory("Subscribed!");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    addToHistory("Failed to subscribe");
                }
            }, new IMqttMessageListener() {
                /**
                 * @desc 訊息到達回撥
                 * @param topic
                 * @param message
                 */
                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    // message Arrived!
                    final String messageRecive = new String(message.getPayload());
                    runOnUiThread(new Runnable() {
                        @Override
                        public void run() {
                            addToHistory("recive message:" + messageRecive);
                        }
                    });
                    System.out.println("Message: " + topic + " : " + messageRecive);

                }
            });

        } catch (MqttException ex) {
            System.err.println("Exception whilst subscribing");
            ex.printStackTrace();
        }
    }

傳送

使用Apollo作為伺服器來擔當訊息代理，又有mqttv3封裝庫，接收只需要訂閱主題，傳送只需往某個主題傳送訊息，在mqttv3中也已經封裝好了。

/**
     * 傳送訊息¬
     */
    private static void publish() {
        //主題
        String topic = "主題";
        //訊息內容
        String content = "來嘛，又給你推送訊息";
        //設定qos
        int qos = 0;
        //伺服器URI
        String broker = "tcp://ip:port"
        //客戶端id
        String clientId = "JavaSample";

        MemoryPersistence persistence = new MemoryPersistence();
        try {
            //傳送依舊使用MqttClient
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            sampleClient.setCallback(new MqttCallback() {
                /**
                 * 連線斷開回調
                 * @param throwable
                 */
                @Override
                public void connectionLost(Throwable throwable) {
                    System.out.println("connectionLost");
                }

                /**
                 * 訊息接收回調，如果只發送，這裡沒關係
                 * @param s
                 * @param mqttMessage
                 * @throws Exception
                 */
                @Override
                public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                    System.out.println("messageArrived:" + s);
                }

                /**
                 * 交付完成回撥。在publish訊息的時候會收到此回撥.
                 * qos:
                 * 0 傳送完則回撥
                 * 1 或 2 會在對方收到時候回撥
                 * @param iMqttDeliveryToken id,可通過id來獲取傳送的訊息內容
                 */
                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    try {
                        System.out.println("deliveryComplete token:" + iMqttDeliveryToken.getMessage().toString());
                    } catch (MqttException e) {
                        e.printStackTrace();
                    }
                }
            });
            //連線操作設定
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setUserName(userName);
            connOpts.setPassword(passWord.toCharArray());
            connOpts.setCleanSession(false);
            System.out.println("Connecting to broker: " + broker);
            sampleClient.connect(connOpts);
            System.out.println("Connected");
            System.out.println("Publishing message: " + content);

            /*傳送，使用MqttMessage封裝類來進行傳送*/
            MqttMessage message = new MqttMessage(content.getBytes());
            //設定優先順序
            message.setQos(qos);
            //設定是否被伺服器保留
            message.setRetained(true);
            //使用mqttClient進行傳送到對應主題
            sampleClient.publish(topic, message);
            if (!sampleClient.isConnected()) {
                System.out.println(mqttAndroidClient.getBufferedMessageCount() + " messages in buffer.");
            }
            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();
        }
    }

• 離線後，可以看到離線後傳送訊息是存在了快取裡，在設定了離線快取配置後，快取大小這些都由自己控制啦，待上線之後再自動進行傳送

paho.mqtt.java.example I/System.out: LOG: Message Published
paho.mqtt.java.example I/System.out: LOG: 2 messages in buffer.
 paho.mqtt.java.example I/System.out: LOG: Message Published
paho.mqtt.java.example I/System.out: LOG: 3 messages in buffer.
paho.mqtt.java.example I/System.out: LOG: Message Published
paho.mqtt.java.example I/System.out: LOG: 4 messages in buffer.
paho.mqtt.java.example I/System.out: LOG: Message Published
paho.mqtt.java.example I/System.out: LOG: 5 messages in buffer.
paho.mqtt.java.example I/System.out: LOG: Message Published
paho.mqtt.java.example I/System.out: LOG: 6 messages in buffer.

設定了最大快取條數，那麼達到條數之後，就不會再存再裡面咯。

• QOS

qos代表服務質量，指的是交通優先順序和資源預留控制機制，而不是接收的服務質量。 服務質量是為不同應用程式，使用者或資料流提供的不同優先順序的能力，或者也可以說是為資料流保證一定的效能水平的能力。
以下是每一個服務質量級別的具體描述
0 ：最多一次傳送 (只負責傳送，傳送過後就不管資料的傳送情況)
1 ： 至少一次傳送 (確認資料交付)
2 ：正好一次傳送 (保證資料交付成功)

demo專案地址：https://github.com/nxSin/paho.mqtt.android。該專案frok自eclipese下專案,如果是Android使用，建議參考這個庫來進行二次封裝使用，內部對其封裝的MqttAndroidClient使用了Service來進行處理連線和收發，完全適用Android場景。