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

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

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

采用Callback式 发布主题

Java代码


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式 订阅主题

Java代码


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();

}

});

}

}

}