1、客户端库下载

如下图所示，有不用编程语言当前支持情况说明。

如下图所示，咱们此处已Java为例，下载正式发布的版本。

当前最新版本为Java最新版本为1.2.2。

下载到的jar包如下图所示：

将该jar包导入到我们的项目中，就可以使用了。

2、登陆连接

先创建MqttClinet对象。

private volatileMqttClient mqttClient;private volatileMqttMessage mqttMessage;privateMqttServerEntity mqttServerEntity;//初始化MQTTClient对象

private voidinitClient() {try{

mqttClient= newMqttClient(getHostUrl(), getClientId());

}catch(MqttException e) {

LogUtils.error(logger, e);

mqttClient= null;

}

}

封装连接参数。

设置回调接口。

准备工作做好后，执行连接即可。

//连接MQTT服务器

public voidstartClient() {

initClient();if (mqttClient == null) {

LogUtils.info(logger,"mqttClient is null");return;

}

MqttConnectOptions options= newMqttConnectOptions();

options.setUserName(mqttServerEntity.getUsername());

options.setPassword(mqttServerEntity.getPassword().toCharArray());

options.setConnectionTimeout(5); //设置超时时间

options.setCleanSession(getCleanSession());

options.setKeepAliveInterval(getKeepAliveInterval());//设置会话心跳时间

options.setAutomaticReconnect(true); //自动重连

try{

mqttClient.setCallback(newBtcMqttCallback());

mqttClient.connect(options);

subscribe();

}catch(Exception e) {

LogUtils.error(logger, e);

}

LogUtils.info(logger,"startClient() isConnected:" +mqttClient.isConnected());

}

3、订阅主题

订阅主题发生在服务器连接登陆成功之后，这里主要有两点，发布消息的服务质量、以及订阅的主题信息。

//订阅主题

private voidsubscribe() {try{int[] Qos ={getQos()};

String[] topic1={mqttServerEntity.getSubscribeTopic()};

mqttClient.subscribe(topic1, Qos);

}catch(Exception e) {

LogUtils.error(logger, e);

}

}

4、发送消息

发送消息时要保证当前客户端与服务器处于连接成功的状态。将主题及消息封装好后，调用发送接口即可。

//发送消息

public voidsendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic) {try{if (mqttMessage == null) {

mqttMessage= newMqttMessage();

mqttMessage.setQos(getQos());

mqttMessage.setRetained(true);}mqttMessage.setPayload(data.getBytes("UTF-8"));mqttClient.publish(topic, mqttMessage);

}catch(Exception e) {

LogUtils.error(logger, e);

}

}

5、消息接收

消息接收是采用回调接口的形式，是建立连接之前设置的，连接成功之后，只有有消息就会回调到下面的方法。

public class BtcMqttCallback implementsMqttCallbackExtended {public voidconnectionLost(Throwable cause) {

LogUtils.info(logger,"connection lost");

}public voiddeliveryComplete(IMqttDeliveryToken token) {

LogUtils.info(logger,"delivery Complete:" +token.isComplete());

}public voidmessageArrived(String topic, MqttMessage message) {

String msg= new String(message.getPayload(), Charset.forName("UTF-8"));

LogUtils.info(logger,"messageArrived() topic:" +topic);

LogUtils.info(logger, msg);

MessageCache.getInstance().putMessage(msg);

}

@Overridepublic void connectComplete(booleanreconnect, String serverURI) {

LogUtils.info(logger,"connectComplete() reconnect:" + reconnect + " serverURI:" +serverURI);

subscribe();

}

}

【参考资料】