MQTT的学习研究（十四） MQTT moquette 的 Callback API 消息发布订阅的实现

   在moquette-mqtt中提供了回调callback模式的发布和订阅但是在订阅之后没有发现有消息接收的方法，参看moquette-mqtt中Block，Future式的发布订阅基础是callback式订阅发布，但是本人在研究源代码测试，发现

callback方式接收没有成功。所以本文中只是callback式的发布和订阅没有消息接收的过程，尚未查到原因。

 

 

采用Callback式 发布主题

package com.etrip.mqtt.callback;

import java.net.URISyntaxException;

import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.Callback;
import org.fusesource.mqtt.client.CallbackConnection;
import org.fusesource.mqtt.client.Listener;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * MQTT moquette 的Server 段用于并发布主题信息
 * 
 * 采用Callback式 发布主题 
 * 
 * @author longgangbai
 */
public class MQTTCallbackServer {
	  private static final Logger LOG = LoggerFactory.getLogger(MQTTCallbackServer.class);
		private final static String CONNECTION_STRING = "tcp://localhost:1883";
		private final static boolean CLEAN_START = true;
		private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s
		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();
	    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
			 final CallbackConnection connection = mqtt.callbackConnection();
			 
			
		    //添加连接的监听事件
		    connection.listener(new Listener() {
		      
		        public void onDisconnected() {
		        }
		        public void onConnected() {
		        }

		        public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) {
		            // You can now process a received message from a topic.
		            // Once process execute the ack runnable.
		            ack.run();
		            System.out.println("topic"+topic.toString()+"="+new String(payload.getData()));
		        }
		        public void onFailure(Throwable value) {
		        }
		    });
		    //添加连接事件
		    connection.connect(new Callback<Void>() {
		    	/**
		    	 * 连接失败的操作
		    	 */
		        public void onFailure(Throwable value) {
		             // If we could not connect to the server.
		        	System.out.println("MQTTCallbackServer.CallbackConnection.connect.onFailure"+"连接失败......"+value.getMessage());
		        	value.printStackTrace();
		        }
		  
		        /**
		         * 连接成功的操作
		         * @param v
		         */
		        public void onSuccess(Void v) {
		        
                 int count=1;
                 while(true){
                	count++;
 		            // 用于发布消息，目前手机段不需要向服务端发送消息
                	//主题的内容
                	final String message="hello "+count+"chinese people !";
 		            final String topic = "china/beijing";
 		            System.out.println("MQTTCallbackServer  publish  topic="+topic+" message :"+message);
					connection.publish(topic, message.getBytes(), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
 		                public void onSuccess(Void v) {
 		                  // the pubish operation completed successfully.
 		                }
 		                public void onFailure(Throwable value) {
 		                	value.printStackTrace();
 		                }
 		            });
 		            try {
						Thread.sleep(2000);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
                 }
		            
//		            //连接断开
//		            connection.disconnect(new Callback<Void>() {
//		                public void onSuccess(Void v) {
//		                  // called once the connection is disconnected.
//		                	System.out.println("MQTTSubscribeClient.CallbackConnection.connect.disconnect.onSuccess", "called once the connection is disconnected.");
//		                }
//		                public void onFailure(Throwable value) {
//		                  // Disconnects never fail.
//		                	System.out.println("MQTTSubscribeClient.CallbackConnection.connect.disconnect.onFailure", "Disconnects never fail."+value.getMessage());
//		                	value.printStackTrace();
//		                }
//		            });
		            
		            
		        }
		    });
		    Thread.sleep(10000000000L);
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally{
			
		}
	}
}

 

采用Callback式 订阅主题

 

package com.etrip.mqtt.callback;

import java.net.URISyntaxException;

import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.Callback;
import org.fusesource.mqtt.client.CallbackConnection;
import org.fusesource.mqtt.client.Listener;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * MQTT moquette 的Client 段用于订阅主题，并接收主题信息
 * 
 * 采用Callback式 订阅主题 
 * 
 * @author longgangbai
 */
public class MQTTCallbackClient {
	  private static final Logger LOG = LoggerFactory.getLogger(MQTTCallbackClient.class);
		private final static String CONNECTION_STRING = "tcp://localhost:1883";
		private final static boolean CLEAN_START = true;
		private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s
		public  static Topic[] topics = {
			    		new Topic("china/beijing", QoS.AT_MOST_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;
		 final String topic = "china/beijing";
		public final static int SEND_BUFFER_SIZE=2*1024*1024;//发送最大缓冲为2M
		
		
		
	  public static void main(String[] args)   {
		//创建MQTT对象
	    MQTT mqtt = new MQTT();
    	//设置mqtt broker的ip和端口
		try {
			mqtt.setHost(CONNECTION_STRING);
		} catch (URISyntaxException e1) {
			e1.printStackTrace();
		}
		//连接前清空会话信息
		mqtt.setCleanSession(CLEAN_START);
		//设置重新连接的次数
		mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);
		//设置重连的间隔时间
		mqtt.setReconnectDelay(RECONNECTION_DELAY);
		//设置心跳时间
		mqtt.setKeepAlive(KEEP_ALIVE);
		//设置缓冲的大小
		mqtt.setSendBufferSize(SEND_BUFFER_SIZE);
	    //获取mqtt的连接对象CallbackConnection
		final CallbackConnection connection= mqtt.callbackConnection();
	    try {
	
		    //添加连接的监听事件
		    connection.listener(new Listener() {
		      
		        public void onDisconnected() {
		        }
		        public void onConnected() {
		        	System.out.println(" 连接成功!");
		        }

		        public void onPublish(UTF8Buffer topic, Buffer payload, Runnable onComplete) {
		        
		        }
		        public void onFailure(Throwable value) {
		        
		        }
		    });
		    //添加连接事件
		    connection.connect(new Callback<Void>() {
		    	/**
		    	 * 连接失败的操作
		    	 */
		        public void onFailure(Throwable value) {
		             // If we could not connect to the server.
		        	System.out.println("MQTTSubscribeClient.CallbackConnection.connect.onFailure  连接失败......"+value.getMessage());
		        	value.printStackTrace();
		        }
		   

		        /**
		         * 连接成功的操作
		         * @param v
		         */
		        public void onSuccess(Void v) {
		          	System.out.println("MQTTSubscribeClient.CallbackConnection.connect.onSuccess 订阅连接成功......");
			          
		            //订阅相关的主题
		            connection.subscribe(topics, new Callback<byte[]>() {
		                public void onSuccess(byte[] qoses) {
		                	System.out.println("MQTTSubscribeClient.CallbackConnection.connect.subscribe.onSuccess 订阅主题成功......");
		                }
		                public void onFailure(Throwable value) {
		                     // subscribe failed.
		                	System.out.println("MQTTSubscribeClient.CallbackConnection.connect.subscribe.onSuccess 订阅主题失败！"+value.getMessage());
		                	value.printStackTrace();
		                }
		            });

		            
		        }
		    });
		    Thread.sleep(100000000000L);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally{
//            //连接断开
            connection.disconnect(new Callback<Void>() {
                public void onSuccess(Void v) {
                  // called once the connection is disconnected.
                	System.out.println("MQTTSubscribeClient.CallbackConnection.connect.disconnect.onSuccess called once the connection is disconnected.");
                }
                public void onFailure(Throwable value) {
                  // Disconnects never fail.
                	System.out.println("MQTTSubscribeClient.CallbackConnection.connect.disconnect.onFailure  Disconnects never fail."+value.getMessage());
                	value.printStackTrace();
                }
            });
		}
	}
}

 

 

评论

2 楼 rongina 2014-07-15  

楼主，activemq下如何在程序中判断Mqtt客户端是否在线呀？

1 楼 ld839431455 2014-07-10  

问题出现在server端的while循环中，把while循环去掉，就可以接受消息