MQTT的學習研究（十四） MQTT moquette 的 Callback API 消息發佈訂閱的實現

時間  2019-11-12
標籤 mqtt 學習 研究 十四 moquette callback api 消息 發佈 訂閱 實現 简体版
原文   原文鏈接

在moquette-mqtt中提供了回調callback模式的發佈和訂閱可是在訂閱以後沒有發現有消息接收的方法，參看moquette-mqtt中Block，Future式的發佈訂閱基礎是callback式訂閱發佈，可是本人在研究源代碼測試，發現java

callback方式接收沒有成功。因此本文中只是callback式的發佈和訂閱沒有消息接收的過程，還沒有查到緣由。網絡

 

 

採用Callback式 發佈主題併發

Java代碼   收藏代碼
  1. package com.etrip.mqtt.callback;  
  2.   
  3. import java.net.URISyntaxException;  
  4.   
  5. import org.fusesource.hawtbuf.Buffer;  
  6. import org.fusesource.hawtbuf.UTF8Buffer;  
  7. import org.fusesource.mqtt.client.Callback;  
  8. import org.fusesource.mqtt.client.CallbackConnection;  
  9. import org.fusesource.mqtt.client.Listener;  
  10. import org.fusesource.mqtt.client.MQTT;  
  11. import org.fusesource.mqtt.client.QoS;  
  12. import org.fusesource.mqtt.client.Topic;  
  13. import org.slf4j.Logger;  
  14. import org.slf4j.LoggerFactory;  
  15.   
  16. /** 
  17.  *  
  18.  * MQTT moquette 的Server 段用於併發布主題信息 
  19.  *  
  20.  * 採用Callback式 發佈主題  
  21.  *  
  22.  * @author longgangbai 
  23.  */  
  24. public class MQTTCallbackServer {  
  25.       private static final Logger LOG = LoggerFactory.getLogger(MQTTCallbackServer.class);  
  26.         private final static String CONNECTION_STRING = "tcp://localhost:1883";  
  27.         private final static boolean CLEAN_START = true;  
  28.         private final static short KEEP_ALIVE = 30;// 低耗網絡，可是又須要及時獲取數據，心跳30s  
  29.         public  static Topic[] topics = {  
  30.                         new Topic("china/beijing", QoS.EXACTLY_ONCE),  
  31.                         new Topic("china/tianjin", QoS.AT_LEAST_ONCE),  
  32.                         new Topic("china/henan", QoS.AT_MOST_ONCE)};  
  33.         public final  static long RECONNECTION_ATTEMPT_MAX=6;  
  34.         public final  static long RECONNECTION_DELAY=2000;  
  35.           
  36.         public final static int SEND_BUFFER_SIZE=2*1024*1024;//發送最大緩衝爲2M  
  37.           
  38.           
  39.       public static void main(String[] args)   {  
  40.         //建立MQTT對象  
  41.         MQTT mqtt = new MQTT();  
  42.         try {  
  43.             //設置mqtt broker的ip和端口  
  44.             mqtt.setHost(CONNECTION_STRING);  
  45.             //鏈接前清空會話信息  
  46.             mqtt.setCleanSession(CLEAN_START);  
  47.             //設置從新鏈接的次數  
  48.             mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);  
  49.             //設置重連的間隔時間  
  50.             mqtt.setReconnectDelay(RECONNECTION_DELAY);  
  51.             //設置心跳時間  
  52.             mqtt.setKeepAlive(KEEP_ALIVE);  
  53.             //設置緩衝的大小  
  54.             mqtt.setSendBufferSize(SEND_BUFFER_SIZE);  
  55.               
  56.               
  57.   
  58.               
  59.             //獲取mqtt的鏈接對象BlockingConnection  
  60.              final CallbackConnection connection = mqtt.callbackConnection();  
  61.                
  62.               
  63.             //添加鏈接的監聽事件  
  64.             connection.listener(new Listener() {  
  65.                 
  66.                 public void onDisconnected() {  
  67.                 }  
  68.                 public void onConnected() {  
  69.                 }  
  70.   
  71.                 public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) {  
  72.                     // You can now process a received message from a topic.  
  73.                     // Once process execute the ack runnable.  
  74.                     ack.run();  
  75.                     System.out.println("topic"+topic.toString()+"="+new String(payload.getData()));  
  76.                 }  
  77.                 public void onFailure(Throwable value) {  
  78.                 }  
  79.             });  
  80.             //添加鏈接事件  
  81.             connection.connect(new Callback<Void>() {  
  82.                 /** 
  83.                  * 鏈接失敗的操做 
  84.                  */  
  85.                 public void onFailure(Throwable value) {  
  86.                      // If we could not connect to the server.  
  87.                     System.out.println("MQTTCallbackServer.CallbackConnection.connect.onFailure"+"鏈接失敗......"+value.getMessage());  
  88.                     value.printStackTrace();  
  89.                 }  
  90.             
  91.                 /** 
  92.                  * 鏈接成功的操做 
  93.                  * @param v 
  94.                  */  
  95.                 public void onSuccess(Void v) {  
  96.                   
  97.                  int count=1;  
  98.                  while(true){  
  99.                     count++;  
  100.                     // 用於發佈消息，目前手機段不須要向服務端發送消息  
  101.                     //主題的內容  
  102.                     final String message="hello "+count+"chinese people !";  
  103.                     final String topic = "china/beijing";  
  104.                     System.out.println("MQTTCallbackServer  publish  topic="+topic+" message :"+message);  
  105.                     connection.publish(topic, message.getBytes(), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {  
  106.                         public void onSuccess(Void v) {  
  107.                           // the pubish operation completed successfully.  
  108.                         }  
  109.                         public void onFailure(Throwable value) {  
  110.                             value.printStackTrace();  
  111.                         }  
  112.                     });  
  113.                     try {  
  114.                         Thread.sleep(2000);  
  115.                     } catch (InterruptedException e) {  
  116.                         // TODO Auto-generated catch block  
  117.                         e.printStackTrace();  
  118.                     }  
  119.                  }  
  120.                       
  121. //                  //鏈接斷開  
  122. //                  connection.disconnect(new Callback<Void>() {  
  123. //                      public void onSuccess(Void v) {  
  124. //                        // called once the connection is disconnected.  
  125. //                          System.out.println("MQTTSubscribeClient.CallbackConnection.connect.disconnect.onSuccess", "called once the connection is disconnected.");  
  126. //                      }  
  127. //                      public void onFailure(Throwable value) {  
  128. //                        // Disconnects never fail.  
  129. //                          System.out.println("MQTTSubscribeClient.CallbackConnection.connect.disconnect.onFailure", "Disconnects never fail."+value.getMessage());  
  130. //                          value.printStackTrace();  
  131. //                      }  
  132. //                  });  
  133.                       
  134.                       
  135.                 }  
  136.             });  
  137.             Thread.sleep(10000000000L);  
  138.         } catch (URISyntaxException e) {  
  139.             // TODO Auto-generated catch block  
  140.             e.printStackTrace();  
  141.         } catch (Exception e) {  
  142.             // TODO Auto-generated catch block  
  143.             e.printStackTrace();  
  144.         }finally{  
  145.               
  146.         }  
  147.     }  
  148. }  

 

採用Callback式 訂閱主題tcp

 

Java代碼   收藏代碼
  1. package com.etrip.mqtt.callback;  
  2.   
  3. import java.net.URISyntaxException;  
  4.   
  5. import org.fusesource.hawtbuf.Buffer;  
  6. import org.fusesource.hawtbuf.UTF8Buffer;  
  7. import org.fusesource.mqtt.client.Callback;  
  8. import org.fusesource.mqtt.client.CallbackConnection;  
  9. import org.fusesource.mqtt.client.Listener;  
  10. import org.fusesource.mqtt.client.MQTT;  
  11. import org.fusesource.mqtt.client.QoS;  
  12. import org.fusesource.mqtt.client.Topic;  
  13. import org.slf4j.Logger;  
  14. import org.slf4j.LoggerFactory;  
  15.   
  16. /** 
  17.  *  
  18.  * MQTT moquette 的Client 段用於訂閱主題，並接收主題信息 
  19.  *  
  20.  * 採用Callback式 訂閱主題  
  21.  *  
  22.  * @author longgangbai 
  23.  */  
  24. public class MQTTCallbackClient {  
  25.       private static final Logger LOG = LoggerFactory.getLogger(MQTTCallbackClient.class);  
  26.         private final static String CONNECTION_STRING = "tcp://localhost:1883";  
  27.         private final static boolean CLEAN_START = true;  
  28.         private final static short KEEP_ALIVE = 30;// 低耗網絡，可是又須要及時獲取數據，心跳30s  
  29.         public  static Topic[] topics = {  
  30.                         new Topic("china/beijing", QoS.AT_MOST_ONCE),  
  31.                         new Topic("china/tianjin", QoS.AT_LEAST_ONCE),  
  32.                         new Topic("china/henan", QoS.AT_MOST_ONCE)};  
  33.         public final  static long RECONNECTION_ATTEMPT_MAX=6;  
  34.         public final  static long RECONNECTION_DELAY=2000;  
  35.          final String topic = "china/beijing";  
  36.         public final static int SEND_BUFFER_SIZE=2*1024*1024;//發送最大緩衝爲2M  
  37.           
  38.           
  39.           
  40.       public static void main(String[] args)   {  
  41.         //建立MQTT對象  
  42.         MQTT mqtt = new MQTT();  
  43.         //設置mqtt broker的ip和端口  
  44.         try {  
  45.             mqtt.setHost(CONNECTION_STRING);  
  46.         } catch (URISyntaxException e1) {  
  47.             e1.printStackTrace();  
  48.         }  
  49.         //鏈接前清空會話信息  
  50.         mqtt.setCleanSession(CLEAN_START);  
  51.         //設置從新鏈接的次數  
  52.         mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);  
  53.         //設置重連的間隔時間  
  54.         mqtt.setReconnectDelay(RECONNECTION_DELAY);  
  55.         //設置心跳時間  
  56.         mqtt.setKeepAlive(KEEP_ALIVE);  
  57.         //設置緩衝的大小  
  58.         mqtt.setSendBufferSize(SEND_BUFFER_SIZE);  
  59.         //獲取mqtt的鏈接對象CallbackConnection  
  60.         final CallbackConnection connection= mqtt.callbackConnection();  
  61.         try {  
  62.       
  63.             //添加鏈接的監聽事件  
  64.             connection.listener(new Listener() {  
  65.                 
  66.                 public void onDisconnected() {  
  67.                 }  
  68.                 public void onConnected() {  
  69.                     System.out.println(" 鏈接成功!");  
  70.                 }  
  71.   
  72.                 public void onPublish(UTF8Buffer topic, Buffer payload, Runnable onComplete) {  
  73.                   
  74.                 }  
  75.                 public void onFailure(Throwable value) {  
  76.                   
  77.                 }  
  78.             });  
  79.             //添加鏈接事件  
  80.             connection.connect(new Callback<Void>() {  
  81.                 /** 
  82.                  * 鏈接失敗的操做 
  83.                  */  
  84.                 public void onFailure(Throwable value) {  
  85.                      // If we could not connect to the server.  
  86.                     System.out.println("MQTTSubscribeClient.CallbackConnection.connect.onFailure  鏈接失敗......"+value.getMessage());  
  87.                     value.printStackTrace();  
  88.                 }  
  89.              
  90.   
  91.                 /** 
  92.                  * 鏈接成功的操做 
  93.                  * @param v 
  94.                  */  
  95.                 public void onSuccess(Void v) {  
  96.                     System.out.println("MQTTSubscribeClient.CallbackConnection.connect.onSuccess 訂閱鏈接成功......");  
  97.                         
  98.                     //訂閱相關的主題  
  99.                     connection.subscribe(topics, new Callback<byte[]>() {  
  100.                         public void onSuccess(byte[] qoses) {  
  101.                             System.out.println("MQTTSubscribeClient.CallbackConnection.connect.subscribe.onSuccess 訂閱主題成功......");  
  102.                         }  
  103.                         public void onFailure(Throwable value) {  
  104.                              // subscribe failed.  
  105.                             System.out.println("MQTTSubscribeClient.CallbackConnection.connect.subscribe.onSuccess 訂閱主題失敗！"+value.getMessage());  
  106.                             value.printStackTrace();  
  107.                         }  
  108.                     });  
  109.   
  110.                       
  111.                 }  
  112.             });  
  113.             Thread.sleep(100000000000L);  
  114.         } catch (Exception e) {  
  115.             // TODO Auto-generated catch block  
  116.             e.printStackTrace();  
  117.         }finally{  
  118. //            //鏈接斷開  
  119.             connection.disconnect(new Callback<Void>() {  
  120.                 public void onSuccess(Void v) {  
  121.                   // called once the connection is disconnected.  
  122.                     System.out.println("MQTTSubscribeClient.CallbackConnection.connect.disconnect.onSuccess called once the connection is disconnected.");  
  123.                 }  
  124.                 public void onFailure(Throwable value) {  
  125.                   // Disconnects never fail.  
  126.                     System.out.println("MQTTSubscribeClient.CallbackConnection.connect.disconnect.onFailure  Disconnects never fail."+value.getMessage());  
  127.                     value.printStackTrace();  
  128.                 }  
  129.             });  
  130.         }  
  131.     }  
  132. }  
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程