springboot整合mqtt的demo

package mqtt.receive; import lombok.extern.slf4j.Slf4j; import mqtt.mqttssss.MqttProperties; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttCallbackExtended; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; /** * @param null: * @return null * @author axian * @description TODO * @date 15:27 */ @Component @Slf4j public class MqttAcceptCallback implements MqttCallbackExtended { private static final Logger logger = LoggerFactory.getLogger(MqttAcceptCallback.class); @Autowired private MqttAcceptClient mqttAcceptClient; @Autowired private MqttProperties mqttProperties; /** * 客户端断开后触发 * * @param throwable */ @Override public void connectionLost(Throwable throwable) { logger.info("连接断开，可以重连"); if (MqttAcceptClient.client == null || !MqttAcceptClient.client.isConnected()) { logger.info("【emqx重新连接】...................................................."); mqttAcceptClient.reconnection(); } } /** * 客户端收到消息触发 * * @param topic 主题 * @param mqttMessage 消息 */ @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { //接收消息 logger.info("【接收的载体消息】:" + mqttMessage.getPayload()); logger.info("【接收消息主题】:" + topic); logger.info("【接收消息Qos】:" + mqttMessage.getQos()); // //将字节数组解码为字符串 // String jsonString = new String(mqttMessage.getPayload()); // //使用JSON解析器解析字符串 // ObjectMapper objectMapper = new ObjectMapper(); // JsonNode jsonNode = objectMapper.readTree(jsonString); //// CsiSmartMeter csiSmartMeter = (CsiSmartMeter) ObjectByteExchangeUtils.byteArrayToObject(mqttMessage.getPayload()); // logger.info("【接收消息内容】:" + jsonNode); // //获取到key，时间戳 // JsonNode data = jsonNode.get("data"); // long time = data.get(0).get("tp").asLong(); // JsonNode point = data.get(0).get("point"); // logger.info("【当前时间戳】:" + time); // if(oConvertUtils.isNotEmpty(map.get(time))) { // JSONObject object = map.get(time); // logger.info("从map中获取一半的对象是" + object); // JSONObject fullObject = getFullObject(object, point); // logger.info("拼接好的完整的对象是" + fullObject + ",此时map的长度" + map.size()); // //构造待存储的csiSmartMeter对象 // CsiSmartMeter csiSmartMeter = generateSmartMeter(fullObject, time); // logger.info("待存储的智能电表对象是" + csiSmartMeter); // csiSmartMeterService.save(csiSmartMeter); // map.remove(time); // logger.info("执行清除操作后，此时map的长度是" + map.size()); // }else { // JSONObject halfObject = getHalfObject(point); // map.put(time,halfObject); // logger.info("拼接一半的对象是" + halfObject); // } } /** * 发布消息成功 * * @param token token */ @Override public void deliveryComplete(IMqttDeliveryToken token) { String[] topics = token.getTopics(); for (String topic : topics) { logger.info("向主题【" + topic + "】发送消息成功！"); } try { MqttMessage message = token.getMessage(); byte[] payload = message.getPayload(); String s = new String(payload, "UTF-8"); logger.info("【接收回调消息内容】:" + s); } catch (Exception e) { logger.error("MqttAcceptCallback deliveryComplete error,message:{}", e.getMessage()); e.printStackTrace(); } } /** * 连接emq服务器后触发 * * @param b * @param s */ @Override public void connectComplete(boolean b, String s) { log.info("============================= 客户端【" + MqttAcceptClient.client.getClientId() + "】连接成功！============================="); // 以/#结尾表示订阅所有以test开头的主题 // 订阅所有机构主题 mqttAcceptClient.subscribe(mqttProperties.getDefaultTopic(), 1); } }