MQTT的學習研究（四）moquette-mqtt 的使用之mqtt Blocking API客戶端訂閱並接收主題信息

在上面兩篇關於mqtt的broker的啓動和mqtt的服務端發佈主題信息以後，咱們客戶端須要訂閱相關的信息並接收相關的主題信息。

package com.etrip.mqtt;

import java.net.URISyntaxException;

import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 *
 * MQTT moquette 的Client 段用於訂閱主題，並接收主題信息
 *
 * 採用阻塞式 訂閱主題
 *
 * @author longgangbai
 */
public class MQTTClient {
      private static final Logger LOG = LoggerFactory.getLogger(MQTTClient.class);
        private final static String CONNECTION_STRING = "tcp://192.168.208.46:1883";
        private final static boolean CLEAN_START = true;
        private final static short KEEP_ALIVE = 30;// 低耗網絡，可是又須要及時獲取數據，心跳30s
        private final static String CLIENT_ID = "publishService";
        public  static Topic[] topics = {
                        new Topic("china/beijing", QoS.EXACTLY_ONCE),
                        new Topic("china/tianjin", QoS.AT_LEAST_ONCE),
                        new Topic("china/henan", QoS.AT_MOST_ONCE)};
        public final  static long RECONNECTION_ATTEMPT_MAX=6;
        public final  static long RECONNECTION_DELAY=2000;

        public final static int SEND_BUFFER_SIZE=2*1024*1024;//發送最大緩衝爲2M


      public static void main(String[] args)   {
        //建立MQTT對象
        MQTT mqtt = new MQTT();
        BlockingConnection connection=null;
        try {
            //設置mqtt broker的ip和端口
            mqtt.setHost(CONNECTION_STRING);
            //鏈接前清空會話信息
            mqtt.setCleanSession(CLEAN_START);
            //設置從新鏈接的次數
            mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);
            //設置重連的間隔時間
            mqtt.setReconnectDelay(RECONNECTION_DELAY);
            //設置心跳時間
            mqtt.setKeepAlive(KEEP_ALIVE);
            //設置緩衝的大小
            mqtt.setSendBufferSize(SEND_BUFFER_SIZE);


            //獲取mqtt的鏈接對象BlockingConnection
            connection = mqtt.blockingConnection();
            //MQTT鏈接的建立
            connection.connect();
            //建立相關的MQTT 的主題列表
            Topic[] topics = {new Topic("china/beijing", QoS.AT_LEAST_ONCE)};
            //訂閱相關的主題信息
            byte[] qoses = connection.subscribe(topics);
            //
            while(true){
                //接收訂閱的消息內容
                Message message = connection.receive();
                //獲取訂閱的消息內容
                byte[] payload = message.getPayload();
                // process the message then:
                LOG.info("MQTTClient Message  Topic="+message.getTopic()+" Content :"+new String(payload));
                //簽收消息的回執
                message.ack();

                Thread.sleep(2000);
            }
        } catch (URISyntaxException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }finally{
            try {
                connection.disconnect();
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}