MQTT的学习研究（六） MQTT moquette 的 Blocking API 订阅消息客户端使用
QiHY 2012-10-17
参阅官方文档：
* 使用 Java 为 MQ Telemetry Transport 创建订户
*在此任务中，您将遵循教程来创建订户应用程序。订户将针对主题创建预订并接收该预订的发布。
*提供了一个示例订户应用程序Subscribe。Subscribe将创建预订主题MQTTExamples，并等待获
*得该预订的发布，等待时间为30秒。订户可以创建预订并等待获得发布。它还可以接收发送至先前
*为同一客户机标识创建的预订的发布。
*MqttConnectionOptions.cleanSessionBoolean属性将控制是否接收到先前所发送的发布
*
*4.创建新的MqttClient实例。
*MqttClientclient=newMqttClient(Example.TCPAddress,Example.clientId);为客户机提供服务器地址，稍后会将此地址用来连接至WebSphereMQ。设置客户机标识以对客户机命名。
*
*◦（可选）可以提供MqttClientPersistence接口的实现以替换缺省实现。缺省MqttPersistence实现会将正在等待传递的QoS1和QoS2消息作为文件来存储；请参阅MQTT客户机中的消息持久性。
*◦MQTT的缺省WebSphereMQTCP/IP端口为1883。对于SSL，缺省端口为8883。在此示例中，缺省地址设置为tcp://localhost:1883。
*◦通常，能够使用客户机标识来标识特定物理客户机很重要。在与服务器相连的所有客户机中，客户机标识必须是唯一的；请参阅MQTT客户机标识。如果与前一个实例使用同一个客户机标识，那么表示目前的实例是同一个客户机的实例。如果您在两个正在运行的客户机中重复使用同一个客户机标识，那么这两个客户机中都会抛出异常，并且一个客户机会终止。
*◦客户机标识的长度不能超过23个字节。如果超过了此长度，就会抛出异常。客户机标识中必须只包含队列管理器名称中允许使用的字符；例如，不能包含连字符或空格。
*◦在您调用MqttClient.connect方法之前，不会处理消息。
*使用客户机对象来发布和预订主题以及恢复有关尚未传递的发布的信息。
*
*
*6.创建一个MqttConnectOptions对象，并设置其cleanSession属性。
*a.创建一个MqttConnectOptions对象。
*MqttConnectOptionsconOptions=newMqttConnectOptions();conOptions是MqttClient构造函数的一个选项参数。
*
*b.设置clearSession属性。
*conOptions.setCleanSession(Example.cleanSession);缺省情况下，Example.cleanSession参数设置为true，从而与MqttConnectionOptions.cleanSession的缺省设置相匹配。
*
*如果您使用缺省MqttConnectOptions，或者在连接客户机之前将MqttConnectOptions.cleanSession设置为true，那么在客户机建立连接时，将除去客户机的任何旧预订。当客户机断开连接时，会除去客户机在会话期间创建的任何新预订。
*
*如果您在连接之前将MqttConnectOptions.cleanSession设置为false，那么客户机创建的任何预订都会被添加至客户机在连接之前就已存在的所有预订。当客户机断开连接时，所有预订仍保持活动状态。
*
*要了解cleanSession属性影响预订的方式，另一种方法就是将它视作模态属性。在其缺省方式cleanSession=true下，客户机仅在会话的作用域内创建预订和接收发布。在另一种方式cleanSession=false下，预订是持久预订。客户机可以连接和断开连接，而其预订保持活动状态。当客户机重新连接时，它将接收任何未传递的发布。在它连接之后，它可以自己修改处于活动状态的预订集。
*
*在连接之前，您必须设置cleanSession方式；在整个会话期间都将保持此方式。要更改此属性的设置，必须将客户机断开连接，然后再重新连接客户机。如果您将方式从使用cleanSession=false更改为cleanSession=true，那么此客户机先前的所有预订以及尚未接收到的任何发布都将被废弃。
*MQTT订阅实现类：
package com.etrip.wsmqtt.client;
import com.ibm.micro.client.mqttv3.MqttClient;
import com.ibm.micro.client.mqttv3.MqttConnectOptions;
/**
*
* 使用 Java 为 MQ Telemetry Transport 创建订户
* 在此任务中，您将遵循教程来创建订户应用程序。订户将针对主题创建预订并接收该预订的发布。
* 提供了一个示例订户应用程序 Subscribe。Subscribe 将创建预订主题 MQTT Examples，并等待获
* 得该预订的发布，等待时间为 30 秒。订户可以创建预订并等待获得发布。它还可以接收发送至先前
* 为同一客户机标识创建的预订的发布。
* @author longgangbai
*/
public class WSMQTTClientSubscribe {
public static void main(String[] args) {
try {
//创建MQTT客户端对象
MqttClient client = new MqttClient(WSMQTTClientConstants.TCPAddress, WSMQTTClientConstants.clientId);
//创建客户端MQTT回调类
WSMQTTClientCallBack callback = new WSMQTTClientCallBack(WSMQTTClientConstants.clientId);
//设置MQTT回调
client.setCallback(callback);
//创建一个连接对象
MqttConnectOptions conOptions = new MqttConnectOptions();
//设置清除会话信息
conOptions.setCleanSession(WSMQTTClientConstants.cleanSession);
//设置超时时间
conOptions.setConnectionTimeout(10000);
//设置会话心跳时间
conOptions.setKeepAliveInterval(20000);
//设置最终端口的通知消息
conOptions.setWill(client.getTopic("LastWillTopic"), "the client will stop !".getBytes(), 1, false);
//连接broker
client.connect(conOptions);
System.out.println("Subscribing to topic \"" + WSMQTTClientConstants.topicString
+ "\" for client instance \"" + client.getClientId()
+ "\" using QoS " + WSMQTTClientConstants.QoS + ". Clean session is "
+ WSMQTTClientConstants.cleanSession);
//订阅相关的主题信息
client.subscribe(WSMQTTClientConstants.topicString, WSMQTTClientConstants.QoS);
System.out.println("Going to sleep for " + WSMQTTClientConstants.sleepTimeout / 1000
+ " seconds");
Thread.sleep(100000000000000l);
//关闭相关的MQTT连接
if(client.isConnected()){
client.disconnect();
}
System.out.println("Finished");
} catch (Exception e) {
e.printStackTrace();
}
}
}MQTT订阅回调类：
package com.etrip.wsmqtt.client;
import com.ibm.micro.client.mqttv3.*;
/**
*
* 消息订阅相关的回调类使用
*
* 必须实现MqttCallback的接口并实现对应的相关接口方法
*
* @author longgangbai
*/
public class WSMQTTClientCallBack implements MqttCallback {
private String instanceData = "";
public WSMQTTClientCallBack(String instance) {
instanceData = instance;
}
public void messageArrived(MqttTopic topic, MqttMessage message) {
try {
System.out.println("Message arrived: \"" + message.toString()
+ "\" on topic \"" + topic.toString() + "\" for instance \""
+ instanceData + "\"");
} catch (Exception e) {
e.printStackTrace();
}
}
public void connectionLost(Throwable cause) {
System.out.println("Connection lost on instance \"" + instanceData
+ "\" with cause \"" + cause.getMessage() + "\" Reason code "
+ ((MqttException)cause).getReasonCode() + "\" Cause \""
+ ((MqttException)cause).getCause() + "\"");
cause.printStackTrace();
}
public void deliveryComplete(MqttDeliveryToken token) {
try {
System.out.println("Delivery token \"" + token.hashCode()
+ "\" received by instance \"" + instanceData + "\"");
} catch (Exception e) {
e.printStackTrace();
}
}
}常量类：
package com.etrip.wsmqtt.client;
/**
*
* 消息订阅消息的常量字段
*
* @author longgangbai
*/
public final class WSMQTTClientConstants {
public static final String TCPAddress = System.getProperty("TCPAddress", "tcp://192.168.208.46:1883");
public static String clientId = String.format("%-23.23s",(System.getProperty("user.name") + "_" + (System.getProperty("clientId", "Subscribe."))).trim()).replace('-', '_');
public static final String topicString = System.getProperty("topicString", "china/beijing");
public static final String publication =System.getProperty("publication", "Hello World " + String.format("%tc", System.currentTimeMillis()));
public static final int quiesceTimeout = Integer.parseInt(System.getProperty("timeout", "10000"));
public static final int sleepTimeout =Integer.parseInt(System.getProperty("timeout", "10000000"));
public static final boolean cleanSession = Boolean.parseBoolean(System.getProperty("cleanSession", "false"));
public static final int QoS = Integer.parseInt(System.getProperty("QoS", "1"));
public static final boolean retained = Boolean.parseBoolean(System.getProperty("retained", "false"));
} 