mqtt-client回调方法简介

使用Fusesource mqtt-client作为mqtt客户包使用，其总共提供了三种API：

1.Blocking API（阻塞式API）

2.Future based API

3.Callback/Continuation（阻塞式API）

其中第三种Callback阻塞式是前两种的基础，可以通过前两种的源码中证明：

public FutureConnection(CallbackConnection next){
receiveFutures = new LinkedList();
receivedFrames = new LinkedList();
this.next = next;
this.next.listener(new Listener() {

public void onConnected()
{
connected = true;
}

public void onDisconnected()
{
connected = false;
}

public void onPublish(UTF8Buffer topic, Buffer payload, Runnable onComplete)
{
getDispatchQueue().assertExecuting();
Message msg = new Message(getDispatchQueue(), topic, payload, onComplete);
if(receiveFutures.isEmpty())
receivedFrames.add(msg);
else
((Promise)receiveFutures.removeFirst()).onSuccess(msg);
}

public void onFailure(Throwable value)
{
getDispatchQueue().assertExecuting();
ArrayList tmp = new ArrayList(receiveFutures);
receiveFutures.clear();
Promise future;
for(Iterator i$ = tmp.iterator(); i$.hasNext(); future.onFailure(value))
future = (Promise)i$.next();

connected = false;
}

final FutureConnection this$0;

{
this$0 = FutureConnection.this;
super();
}
}
);
}

毫无疑问Callback方式是最复杂的一种，但是其也是能够提供更好的服务，因此有必要好好研究，下面就是对使用回调方式的简单介绍：

一、在使用回调方式前，先通过MQTT获取回调连接：

MQTT mqtt=new MQTT();
//此处省略设置MQTT的属性
CallbackConnection connection=mqtt.callbackConnection();

二、调用CallbackConnection的方法实现业务逻辑：

1、connect（连接方法）：

connection.connect(new Callback<Void>() {
public void onSuccess(Void arg0) {
//进入该方法表示连接成功连接成功
}
public void onFailure(Throwable value) {
//进入该方法表示连接失败
}
});

一般可以在connect的onSuccess方法中发布或者订阅相应的主题，在其onFailure方法中作相应的断开连接等操作

2、listener（监听方法）：

connection.listener(new Listener() {
@Override
public void onPublish(UTF8Buffer topicmsg, Buffer msg, Runnable ack) {
//utf-8 is used for dealing with the garbled
//String topic = topicmsg.utf8().toString();
//String payload = msg.utf8().toString();
//表示监听成功
ack.run();
}
@Override
public void onFailure(Throwable value) {
//表示监听失败
}
//execute only once when connection is ended
@Override
public void onDisconnected() {
/表示监听到断开连接
}
//execute only once when connecting started
@Override
public void onConnected() {
//表示监听到连接成功
}
});

onPublish表示成功，可以获取到订阅的主题和订阅的内容（UTF8Buffer topicmsg表示订阅的主题, Buffer msg表示订阅的类容），一般的可以在这个方法中获取到订阅的主题和内容然后进行相应的判断和其他业务逻辑操作；

onFailure表示监听失败，这里可以调用相应的断开连接等方法；

onConnected表示监听到连接建立，该方法只在建立连接成功时执行一次，表示连接成功建立，如果有必要可以在该方法中进行相应的订阅操作；

onDisconnected表示监听到连接断开，该方法只在断开连接时执行一次，如有必要可以进行相应的资源回收操作。

3、subscribe（订阅方法）：

connection.subscribe(topics, new Callback<byte[]>() {
@Override
public void onSuccess(byte[] qoses) {
//主题订阅成功
}
@Override
public void onFailure(Throwable value) {
//状态主题订阅失败
}
});

onSuccess方法表示订阅成功，onFailure方法表示订阅失败；方法中Topic[] topics表示定于的主题数组，注意只有在改方法订阅的主题，才能够在监听方法中接收到。

4、publish（发布主题方法）：

connection.publish(topic, payload.getBytes(), qos, retain, new Callback<Void>() {
@Override
public void onSuccess(Void arg0) {
//表示发布主题成功
}
@Override
public void onFailure(Throwable throwable) {
//表示发布主题失败
}
});

publish方法用于发布响应的主题，以便订阅者订阅；onSuccess表示发布成功，onFailure表示发布失败。

5、disconnect（断开连接方法）：

connection.disconnect(new Callback<Void>() {
@Override
public void onSuccess(Void arg0) {
//与服务器断开连接成功
}
@Override
public void onFailure(Throwable arg0) {
//与服务器断开连接失败
}
});

disconnect表示断开与代理服务器的连接，调用该方法后只是表示断开连接但是实际的connect依然存在并没有为null，再次调用connect方法又能够连接成功。

注意：

1、在上面的五个方法中一般先调用connect和listener方法

2、回调将执行与连接相关联的调度队列，以便可以安全使用从回调的连接，但你绝不能在回调中执行任何阻塞操作，否则会改变执行的顺序，这样可能出错。如果可能存在阻塞时，最好是在连接的调度队列中执行另外一个线程：

connection.getDispatchQueue().execute(new Runnable() {
public void run() {
//在这里进行相应的订阅、发布、停止连接等等操作
}
});