码迷,mamicode.com
首页 > Windows程序 > 详细

MQTT的学习研究(四)moquette-mqtt 的使用之mqtt Blocking API客户端订阅并接收主题信息

时间:2015-07-01 18:09:38      阅读:433      评论:0      收藏:0      [点我收藏+]

标签:

在上面两篇关于mqtt的broker的启动和mqtt的服务端发布主题信息之后,我们客户端需要订阅相关的信息并接收相关的主题信息。

Java代码  技术分享
  1. package com.etrip.mqtt;  
  2.   
  3. import java.net.URISyntaxException;  
  4.   
  5. import org.fusesource.mqtt.client.BlockingConnection;  
  6. import org.fusesource.mqtt.client.MQTT;  
  7. import org.fusesource.mqtt.client.Message;  
  8. import org.fusesource.mqtt.client.QoS;  
  9. import org.fusesource.mqtt.client.Topic;  
  10. import org.slf4j.Logger;  
  11. import org.slf4j.LoggerFactory;  
  12. /** 
  13.  *  
  14.  * MQTT moquette 的Client 段用于订阅主题,并接收主题信息 
  15.  *  
  16.  * 采用阻塞式 订阅主题  
  17.  *  
  18.  * @author longgangbai 
  19.  */  
  20. public class MQTTClient {  
  21.       private static final Logger LOG = LoggerFactory.getLogger(MQTTClient.class);  
  22.         private final static String CONNECTION_STRING = "tcp://192.168.208.46:1883";  
  23.         private final static boolean CLEAN_START = true;  
  24.         private final static short KEEP_ALIVE = 30;// 低耗网络,但是又需要及时获取数据,心跳30s  
  25.         private final static String CLIENT_ID = "publishService";  
  26.         public  static Topic[] topics = {  
  27.                         new Topic("china/beijing", QoS.EXACTLY_ONCE),  
  28.                         new Topic("china/tianjin", QoS.AT_LEAST_ONCE),  
  29.                         new Topic("china/henan", QoS.AT_MOST_ONCE)};  
  30.         public final  static long RECONNECTION_ATTEMPT_MAX=6;  
  31.         public final  static long RECONNECTION_DELAY=2000;  
  32.           
  33.         public final static int SEND_BUFFER_SIZE=2*1024*1024;//发送最大缓冲为2M  
  34.           
  35.           
  36.       public static void main(String[] args)   {  
  37.         //创建MQTT对象  
  38.         MQTT mqtt = new MQTT();  
  39.         BlockingConnection connection=null;  
  40.         try {  
  41.             //设置mqtt broker的ip和端口  
  42.             mqtt.setHost(CONNECTION_STRING);  
  43.             //连接前清空会话信息  
  44.             mqtt.setCleanSession(CLEAN_START);  
  45.             //设置重新连接的次数  
  46.             mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);  
  47.             //设置重连的间隔时间  
  48.             mqtt.setReconnectDelay(RECONNECTION_DELAY);  
  49.             //设置心跳时间  
  50.             mqtt.setKeepAlive(KEEP_ALIVE);  
  51.             //设置缓冲的大小  
  52.             mqtt.setSendBufferSize(SEND_BUFFER_SIZE);  
  53.               
  54.               
  55.             //获取mqtt的连接对象BlockingConnection  
  56.             connection = mqtt.blockingConnection();  
  57.             //MQTT连接的创建   
  58.             connection.connect();  
  59.             //创建相关的MQTT 的主题列表   
  60.             Topic[] topics = {new Topic("china/beijing", QoS.AT_LEAST_ONCE)};  
  61.             //订阅相关的主题信息   
  62.             byte[] qoses = connection.subscribe(topics);  
  63.             //  
  64.             while(true){  
  65.                 //接收订阅的消息内容  
  66.                 Message message = connection.receive();  
  67.                 //获取订阅的消息内容   
  68.                 byte[] payload = message.getPayload();  
  69.                 // process the message then:  
  70.                 LOG.info("MQTTClient Message  Topic="+message.getTopic()+" Content :"+new String(payload));  
  71.                 //签收消息的回执  
  72.                 message.ack();  
  73.                   
  74.                 Thread.sleep(2000);  
  75.             }  
  76.         } catch (URISyntaxException e) {  
  77.             // TODO Auto-generated catch block  
  78.             e.printStackTrace();  
  79.         } catch (Exception e) {  
  80.             // TODO Auto-generated catch block  
  81.             e.printStackTrace();  
  82.         }finally{  
  83.             try {  
  84.                 connection.disconnect();  
  85.             } catch (Exception e) {  
  86.                 // TODO Auto-generated catch block  
  87.                 e.printStackTrace();  
  88.             }  
  89.         }  
  90.     }  
  91. }  

 

MQTT的学习研究(四)moquette-mqtt 的使用之mqtt Blocking API客户端订阅并接收主题信息

标签:

原文地址:http://www.cnblogs.com/yudar/p/4613732.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!