MQTT客户端发布订阅

 订阅多个主题

package com.xw.elevator.platform.mqtt;import java.util.concurrent.ScheduledExecutorService;import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;&＃64;Component
public class Client implements CommandLineRunner {private static final Logger logger &＃61; LoggerFactory.getLogger(PushCallback.class);public static final String HOST &＃61; "tcp://192.168.121.36";private static final String clientid &＃61; "client124";public static String Issue_Topic &＃61; "Fatri/Devs/{DeviceNo}/Issue";public static final String DevInfo_Topic &＃61; "Fatri/Devs/Report/DevInfo";public static final String EleStatus_Topic &＃61; "Fatri/Devs/Report/EleStatus";public static final String EleWran_Topic &＃61; "Fatri/Devs/Report/EleWarn";static MqttClient client;private MqttConnectOptions options;private String userName &＃61; "admin";private String passWord &＃61; "123456";private ScheduledExecutorService scheduler;MqttMessage message;volatile static MqttTopic topic;MqttTopic topicD;MqttTopic topicR;static {try {client &＃61; new MqttClient(HOST, clientid, new MemoryPersistence());topic &＃61; client.getTopic(Issue_Topic);} catch (MqttException e) {e.printStackTrace();}}public void connect() {options &＃61; new MqttConnectOptions();options.setCleanSession(false);options.setUserName(userName);options.setPassword(passWord.toCharArray());// 设置超时时间options.setConnectionTimeout(10);// 设置会话心跳时间options.setKeepAliveInterval(20);try {client.setCallback(new PushCallback());client.connect(options);topic &＃61; client.getTopic(Issue_Topic);} catch (Exception e) {e.printStackTrace();}}public void start() {try {int[] Qos &＃61; new int[]{0, 0, 0};String[] topics &＃61; new String[]{DevInfo_Topic, EleStatus_Topic, EleWran_Topic};client.subscribe(topics, Qos);} catch (Exception e) {e.printStackTrace();}}public void publish(MqttTopic topic, MqttMessage message) throws MqttPersistenceException, MqttException {MqttDeliveryToken token &＃61; topic.publish(message);token.waitForCompletion();System.out.println("Client message is published completely! " &＃43; token.isComplete());}&＃64;Overridepublic void run(String... args) throws Exception {Client client &＃61; new Client();client.connect();client.start();logger.info("客户端连接成功");}
}

回调函数

package com.xw.elevator.platform.mqtt;import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.client.RestTemplate;import net.sf.json.JSONObject;import com.alibaba.fastjson.JSON;
import com.xw.elevator.platform.biz.dto.Feature;
import com.xw.elevator.platform.biz.dto.ReveiveElevatorInfoDTO;
import com.xw.elevator.platform.web.ResponseOfReveiveElevatorInfo;/*** 发布消息的回调类*

* 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。* 每个客户机标识都需要一个回调实例。在此示例中&＃xff0c;构造函数传递客户机标识以另存为实例数据。 在回调中&＃xff0c;将它用来标识已经启动了该回调的哪个实例。* 必须在回调类中实现三个方法&＃xff1a;*

* public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。*

* public void connectionLost(Throwable cause)在断开连接时调用。*

* public void deliveryComplete(MqttDeliveryToken token)) 接收到已经发布的 QoS 1 或 QoS 2* 消息的传递令牌时调用。 由 MqttClient.connect 激活此回调。*/
public class PushCallback implements MqttCallback {private static final Logger logger &＃61; LoggerFactory.getLogger(PushCallback.class);private static int number &＃61; 0;// 唯一数字,集群第一台&＃61;0&＃xff0c;第二台&＃61;200000,第三台&＃61;400000private static int maxNum &＃61; 200000;// 最大值,集群第一台&＃61;200000&＃xff0c;第二台&＃61;400000,第三台&＃61;600000private static SimpleDateFormat sdf &＃61; new SimpleDateFormat("yyyyMMddHHmmssSSS");// 年月日格式MqttTopic topic1;public void connectionLost(Throwable cause) {// 连接丢失后&＃xff0c;一般在这里面进行重连logger.info("连接断开&＃xff0c;可以做重连");Client client &＃61; new Client();client.connect();client.start();logger.info("客户端重新连接成功");}public void deliveryComplete(IMqttDeliveryToken token) {logger.info("deliveryComplete---------" &＃43; token.isComplete());}public void publish(MqttTopic topic, MqttMessage message) throws MqttPersistenceException, MqttException {MqttDeliveryToken token &＃61; topic.publish(message);token.waitForCompletion();logger.info("message is published completely! " &＃43; token.isComplete());}public void messageArrived(String topic, MqttMessage message) throws Exception {// subscribe后得到的消息会执行到这里面logger.info("接收消息主题:{}", topic);logger.info("接收消息Qos:{}", message.getQos());logger.info("接收消息内容:{}", new String(message.getPayload()));JSONObject jsonObject &＃61; JSONObject.fromObject(new String(message.getPayload()));logger.info("jsonObject:" &＃43; jsonObject);String DeviceNo &＃61; jsonObject.getString("DeviceNo");String MessageType &＃61; jsonObject.getString("MessageType");// 获取当前时间的字符串SimpleDateFormat sdf &＃61; new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");String date &＃61; sdf.format(new Date());System.out.println("MessageType:" &＃43; MessageType);// 1.接收心跳报文不用应答if (null !&＃61; MessageType && MessageType.equals("AliveInfo")) {logger.info("接收的是心跳报文");return;}logger.info("DeviceNo:" &＃43; DeviceNo);}
}