mqtt client api: 阻塞API

fusesource版本：mqtt-client-1.11.jar
下載地址：https://github.com/fusesource/mqtt-client

fusesource提供三種mqtt client api: 阻塞API，基於Futur的API和回調API。其中，回調API是最復雜的也是性能最好的，另外兩種均是對回調API的封裝。 我們下面就簡單介紹一下回調API的使用方法。

import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.hawtdispatch.Dispatch;
import org.fusesource.hawtdispatch.DispatchQueue;
import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.Callback;
import org.fusesource.mqtt.client.CallbackConnection;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.Listener;
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.fusesource.mqtt.client.Tracer;
import org.fusesource.mqtt.codec.MQTTFrame; 
public class MqttClient {
public static void main(String[] args)
{
 try {
   MQTT mqtt=new MQTT();
   
   //MQTT設置說明
   mqtt.setHost("tcp://10.1.58.191:1883");
   mqtt.setClientId("876543210"); //用於設置客戶端會話的ID。在setCleanSession(false);被調用時，MQTT服務器利用該ID獲得相應的會話。此ID應少於23個字符，默認根據本機地址、端口和時間自動生成
   mqtt.setCleanSession(false); //若設為false，MQTT服務器將持久化客戶端會話的主體訂閱和ACK位置，默認為true
   mqtt.setKeepAlive((short) 60);//定義客戶端傳來消息的最大時間間隔秒數，服務器可以據此判斷與客戶端的連接是否已經斷開，從而避免TCP/IP超時的長時間等待
   mqtt.setUserName("admin");//服務器認證用戶名
   mqtt.setPassword("admin");//服務器認證密碼
   
   mqtt.setWillTopic("willTopic");//設置“遺囑”消息的話題，若客戶端與服務器之間的連接意外中斷，服務器將發布客戶端的“遺囑”消息
   mqtt.setWillMessage("willMessage");//設置“遺囑”消息的內容，默認是長度為零的消息
   mqtt.setWillQos(QoS.AT_LEAST_ONCE);//設置“遺囑”消息的QoS，默認為QoS.ATMOSTONCE
   mqtt.setWillRetain(true);//若想要在發布“遺囑”消息時擁有retain選項，則為true
   mqtt.setVersion("3.1.1");
   
   //失敗重連接設置說明
   mqtt.setConnectAttemptsMax(10L);//客戶端首次連接到服務器時，連接的最大重試次數，超出該次數客戶端將返回錯誤。-1意為無重試上限，默認為-1
   mqtt.setReconnectAttemptsMax(3L);//客戶端已經連接到服務器，但因某種原因連接斷開時的最大重試次數，超出該次數客戶端將返回錯誤。-1意為無重試上限，默認為-1
   mqtt.setReconnectDelay(10L);//首次重連接間隔毫秒數，默認為10ms
   mqtt.setReconnectDelayMax(30000L);//重連接間隔毫秒數，默認為30000ms
   mqtt.setReconnectBackOffMultiplier(2);//設置重連接指數回歸。設置為1則停用指數回歸，默認為2
   
   //Socket設置說明
         mqtt.setReceiveBufferSize(65536);//設置socket接收緩沖區大小，默認為65536（64k）
         mqtt.setSendBufferSize(65536);//設置socket發送緩沖區大小，默認為65536（64k）
         mqtt.setTrafficClass(8);//設置發送數據包頭的流量類型或服務類型字段，默認為8，意為吞吐量最大化傳輸
         
         //帶寬限制設置說明
         mqtt.setMaxReadRate(0);//設置連接的最大接收速率，單位為bytes/s。默認為0，即無限制
         mqtt.setMaxWriteRate(0);//設置連接的最大發送速率，單位為bytes/s。默認為0，即無限制
         
         //選擇消息分發隊列
         mqtt.setDispatchQueue(Dispatch.createQueue("foo"));//若沒有調用方法setDispatchQueue，客戶端將為連接新建一個隊列。如果想實現多個連接使用公用的隊列，顯式地指定隊列是一個非常方便的實現方法
   
         //設置跟蹤器
   mqtt.setTracer(new Tracer(){
             @Override
             public void onReceive(MQTTFrame frame) {
                 System.out.println("recv: "+frame);
             } 
             @Override
             public void onSend(MQTTFrame frame) {
                 System.out.println("send: "+frame);
             } 
             @Override
             public void debug(String message, Object... args) {
                 System.out.println(String.format("debug: "+message, args));
             }
         });
   
   
   
         //使用回調式API
   final CallbackConnection callbackConnection=mqtt.callbackConnection();
   
   //連接監聽
   callbackConnection.listener(new Listener() {
   
   //接收訂閱話題發布的消息
   @Override
   public void onPublish(UTF8Buffer topic, Buffer payload, Runnable onComplete) {
    System.out.println("=============receive msg================"+new String(payload.toByteArray()));
     onComplete.run();
   }
   
   //連接失敗
   @Override
   public void onFailure(Throwable value) {
    System.out.println("===========connect failure===========");
    callbackConnection.disconnect(null);
   }
   
      //連接斷開
   @Override
   public void onDisconnected() {
    System.out.println("====mqtt disconnected=====");
    
   }
   
   //連接成功
   @Override
   public void onConnected() {
    System.out.println("====mqtt connected=====");
    
   }
  });
   
   
   
   //連接
   callbackConnection.connect(new Callback() {
    
     //連接失敗
       public void onFailure(Throwable value) {
           System.out.println("============連接失敗："+value.getLocalizedMessage()+"============");
       } 
       // 連接成功
       public void onSuccess(Void v) { 
           //訂閱主題
           Topic[] topics = {new Topic("foo", QoS.AT_LEAST_ONCE)};
           callbackConnection.subscribe(topics, new Callback<byte[]>() {
               //訂閱主題成功
            public void onSuccess(byte[] qoses) {
                   System.out.println("========訂閱成功=======");
               }
            //訂閱主題失敗
               public void onFailure(Throwable value) {
                System.out.println("========訂閱失敗=======");
                callbackConnection.disconnect(null);
               }
           });
           
           
            //發布消息
           callbackConnection.publish("foo", ("Hello ").getBytes(), QoS.AT_LEAST_ONCE, true, new Callback() {
               public void onSuccess(Void v) {
                 System.out.println("===========消息發布成功============");
               }
               public void onFailure(Throwable value) {
                System.out.println("========消息發布失敗=======");
                callbackConnection.disconnect(null);
               }
           }); 
  
       }
   });
   
   
   
   while(true)
   {
    
   }
 
  
 } catch (Exception e) {
  e.printStackTrace();
 } 
  
}
}