springboot集成mqtt 单topic和多topic

package org.springblade.reservoir.data.acess.gnss.mqtt; import com.alibaba.fastjson.JSONObject; import com.alibaba.nacos.shaded.com.google.gson.Gson; import com.baomidou.mybatisplus.core.toolkit.IdWorker; import com.fasterxml.jackson.core.JsonProcessingException; import org.eclipse.paho.client.mqttv3.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springblade.core.cache.utils.CacheUtil; import org.springblade.core.tool.constant.BladeConstant; import org.springblade.core.tool.utils.DateUtil; import org.springblade.core.tool.utils.Func; import org.springblade.reservior.biz.entity.GnssData; import org.springblade.reservoir.data.acess.gnss.entity.GnssRawData; import org.springblade.reservoir.data.acess.gnss.mapper.GnssDataMapper; import org.springblade.reservoir.data.acess.gnss.mapper.GnssRawDataMapper; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import javax.annotation.Resource; import java.io.UnsupportedEncodingException; import java.time.LocalDateTime; import java.util.Base64; import java.util.Date; import java.util.List; import java.util.Map; /** * @author: hejingping * @Description: 回调类 */ @Component public class MqttCallback implements MqttCallbackExtended { @Value("${mqtt.host}") String host; @Value("${mqtt.username}") String username; @Value("${mqtt.password}") String password; @Value("${mqtt.clientId}") String clientId; private static final String PARANUM_Q = "11"; private static final String PARANUM_PZTBWL = "12"; private static final String PARANUM_X = "13"; private static final String PARANUM_Y = "14"; private static final String PARANUM_Z = "15"; private static final Logger log = LoggerFactory.getLogger(MqttCallback.class); private MqttClient mqttClient; public MqttCallback(MqttClient mqttClient) { this.mqttClient = mqttClient; } @Resource private GnssRawDataMapper gnssRawDataMapper; @Resource private GnssDataMapper gnssDataMapper; public static MqttCallback mqttCallback; //通过@PostConstruct实现初始化bean之前进行的操作 @PostConstruct public void init() { mqttCallback = this; mqttCallback.gnssRawDataMapper = this.gnssRawDataMapper; mqttCallback.gnssDataMapper = this.gnssDataMapper; } /** * 丢失连接，可在这里做重连 * 只会调用一次 * * @param throwable */ @Override public void connectionLost(Throwable throwable) { log.error("mqtt connectionLost 连接断开，5S之后尝试重连: {}", throwable.getMessage()); long reconnectTimes = 1; while (true) { try { if (MqttClient.getClient().isConnected()) { //判断已经重新连接成功 需要重新订阅主题 可以在这个if里面订阅主题 或者 connectComplete（方法里面） 看你们自己选择 log.warn("mqtt reconnect success end 重新连接 重新订阅成功"); return; } reconnectTimes+=1; log.warn("mqtt reconnect times = {} try again... mqtt重新连接时间 {}", reconnectTimes, reconnectTimes); MqttClient.getClient().reconnect(); } catch (MqttException e) { log.error("mqtt断连异常", e); } try { Thread.sleep(5000); } catch (InterruptedException e1) { } } } /** * @param topic * @param mqttMessage * @throws Exception * subscribe后得到的消息会执行到这里面 * 订阅者收到消息之后执行 */ @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { System.out.println("客户端接收到消息之后调用"); insertGnssRawData(mqttMessage); log.info("接收消息主题 : {}，接收消息内容 : {}", topic, mqttMessage); } private void insertGnssRawData(MqttMessage mqttMessage) throws JsonProcessingException { JSONObject jsonObject = JSONObject.parseObject(mqttMessage.toString()); String payload = decode(jsonObject.getString("payload")); String mess = payload.substring(payload.indexOf("{")); JSONObject messObject = JSONObject.parseObject(mess); String stcd = messObject.keySet().iterator().next(); GnssRawData gnssRawData = new GnssRawData(); gnssRawData.setId(IdWorker.getId()); gnssRawData.setTenantId(BladeConstant.ADMIN_TENANT_ID); gnssRawData.setMessage(mess); gnssRawData.setMqttMessage(jsonObject.toString()); gnssRawData.setStcd(stcd); gnssRawData.setCreateTime(DateUtil.now()); gnssRawData.setIsDeleted(BladeConstant.DB_NOT_DELETED); mqttCallback.gnssRawDataMapper.insert(gnssRawData); insertGnssData(jsonObject.getString("messageId"), gnssRawData.getCreateTime(), mess); } private void insertGnssData(String cacheValue,Date now,String mess) throws JsonProcessingException { Object value = CacheUtil.get("id", "type-", "messageId"); if(Func.isNotEmpty(value)){ if(!value.toString().equals(cacheValue)){ CacheUtil.clear("id"); insertData(now, cacheValue,mess); } }else { insertData(now, cacheValue,mess); } } private void insertData(Date now, String cacheValue, String mess) { Gson gson = new Gson(); GnssData gnssData = new GnssData(); Map<String, Map<String, Map<String, List<Map<String, String>>>>> messMap = gson.fromJson(mess, Map.class); String dvcd = messMap.keySet().iterator().next(); Map<String, Map<String, List<Map<String, String>>>> valueMap1 = messMap.get(dvcd); String time = valueMap1.keySet().iterator().next(); Map<String, List<Map<String, String>>> valueMap2 = valueMap1.get(time); String mpcd = valueMap2.keySet().iterator().next(); List<Map<String, String>> valuesList = valueMap2.get(mpcd); for (Map<String, String> map : valuesList) { String paranum = map.get("paranum"); String value = map.get("value"); switch (paranum) { case PARANUM_Q: gnssData.setQ(value); break; case PARANUM_PZTBWL: gnssData.setPztbwl(value); break; case PARANUM_X: gnssData.setX(value); break; case PARANUM_Y: gnssData.setY(value); break; case PARANUM_Z: gnssData.setZ(value); break; default: break; } } gnssData.setId(IdWorker.getId()); gnssData.setTenantId(BladeConstant.ADMIN_TENANT_ID); gnssData.setDvcd(dvcd); gnssData.setMpcd(mpcd); gnssData.setMstm(LocalDateTime.now()); gnssData.setCreateTime(now); gnssData.setIsDeleted(BladeConstant.DB_NOT_DELETED); mqttCallback.gnssDataMapper.insert(gnssData); CacheUtil.put("id", "type-", "messageId", cacheValue); } public static String decode(String base64Str) { // 解码后的字符串 String str = ""; // 解码 byte[] base64Data = Base64.getDecoder().decode(base64Str); try { // byte[]-->String str = new String(base64Data, "utf-8"); } catch (UnsupportedEncodingException