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

 MQTT moquette 的Server發佈主題

package com.etrip.mqtt.future;

import java.net.URISyntaxException;

import org.fusesource.mqtt.client.FutureConnection;
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;

/**
 *
 *
 *
 * 採用Future式 發佈主題
 *
 * @author longgangbai
 */
public class MQTTFutureServer {
        private static final Logger LOG = LoggerFactory.getLogger(MQTTFutureServer.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
        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 = new MQTT();
            try {
                //設置服務端的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);

                //建立鏈接
                final FutureConnection connection= mqtt.futureConnection();
                connection.connect();
                int count=1;
                while(true){
                    count++;
                    // 用於發佈消息，目前手機段不須要向服務端發送消息
                    //主題的內容
                    String message="hello "+count+"chinese people !";
                    String topic = "china/beijing";
                    connection.publish(topic, message.getBytes(), QoS.AT_LEAST_ONCE,
                            false);
                    System.out.println("MQTTFutureServer.publish Message "+"Topic Title :"+topic+" context :"+message);

                }
            } catch (URISyntaxException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
}

 

 

 MQTT moquette 的Client接收主題

 

package com.etrip.mqtt.future;

import java.net.URISyntaxException;

import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
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 段用於訂閱主題，並接收主題信息
 *
 * 採用Future 式 訂閱主題
 *
 * @author longgangbai
 */
public class MQTTFutureClient {
        private static final Logger LOG = LoggerFactory.getLogger(MQTTFutureClient.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();
                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 FutureConnection connection= mqtt.futureConnection();
                    connection.connect();
                    connection.subscribe(topics);
                    while(true){
                        Future<Message> futrueMessage=connection.receive();
                        Message message =futrueMessage.await();


                        System.out.println("MQTTFutureClient.Receive Message "+ "Topic Title :"+message.getTopic()+" context :"+String.valueOf(message.getPayloadBuffer()));
                    }
                } catch (URISyntaxException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } catch (Exception e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }finally{

                }
            }
}