修改消息组件

common-lib/src/main/java/com/mortals/xhx/queue/ConsumerService.java

@@ -56,30 +56,6 @@ public class ConsumerService {
 
                     for (TbQueueMsg item : msgs) {
                         //todo
-//                        Message innerMsg = new Message();
-//                        TbQueueMsgHeaders headMap = item.getHeaders();
-//                        int messageProtocl = headMap.get(MessageHeader.MESSAGEPROTOCOL)[0];
-//                        Map<String, Object> headers = new HashMap<>();
-//                        innerMsg.setPayload(item.getData());
-//                        headers.put(MessageHeader.MESSAGEPROTOCOL, messageProtocl);
-//                        innerMsg.setStoreTime(DateUtils.getCurrDate().getTime());
-//                        if(messageProtocl== MsgTypeEnum.MSG_SYSTEM.getValue()){
-//                            innerMsg.setHeaders(headers);
-//                            sendSysMessage2Cluster(innerMsg);
-//                        }else{
-//                            String clientId = MixAll.convertByteS2Str(headMap.get(MessageHeader.CLIENTID));
-//                            int messageType = headMap.get(MessageHeader.MESSAGETYPE)[0];
-//                            String topic = MixAll.convertByteS2Str(headMap.get(MessageHeader.TOPIC));
-//                            int qos = headMap.get(MessageHeader.QOS)[0];
-//
-//                            innerMsg.setClientId(clientId);
-//                            innerMsg.setType(Message.Type.valueOf(messageType));
-//                            headers.put(MessageHeader.TOPIC, topic);
-//                            headers.put(MessageHeader.QOS, qos);
-//                            headers.put(MessageHeader.RETAIN, false);
-//                            headers.put(MessageHeader.DUP, false);
-//                            innerMsg.setHeaders(headers);
-//                            sendContrlMessage2Cluster(innerMsg);
 //                        }
                     }
                 } catch (Exception e) {

common-lib/src/main/java/com/mortals/xhx/queue/TopicPartitionInfo.java

@@ -8,16 +8,29 @@ import java.util.Optional;
 
 @Data
 public class TopicPartitionInfo {
-    private  String topic;
-    private  Integer partition;
-
-    private  String fullTopicName;
+    /**
+     * topic名称
+     */
+    private String topic;
+    /**
+     * 分区，kafka存在
+     */
+    private Integer partition;
+    /**
+     * 交换机名称，rabbmitmq存在
+     */
+    private String exchangeName;
+    /**
+     * 带分区的topic
+     */
+    private String fullTopicName;
 
 
     @Builder
-    public TopicPartitionInfo(String topic, Integer partition) {
+    public TopicPartitionInfo(String topic, Integer partition, String exchangeName) {
         this.topic = topic;
         this.partition = partition;
+        this.exchangeName = exchangeName;
         String tmp = topic;
         if (partition != null) {
             tmp += "." + partition;
@@ -26,7 +39,7 @@ public class TopicPartitionInfo {
     }
 
     public TopicPartitionInfo newByTopic(String topic) {
-        return new TopicPartitionInfo(topic, this.partition);
+        return new TopicPartitionInfo(topic, this.partition, "");
     }
 
     public String getTopic() {

common-lib/src/main/java/com/mortals/xhx/queue/kafka/AbstractTbQueueConsumerTemplate.java

 package com.mortals.xhx.queue.kafka;
 
 
+import com.alibaba.fastjson.JSON;
 import com.mortals.xhx.queue.TbQueueConsumer;
 import com.mortals.xhx.queue.TbQueueMsg;
 import com.mortals.xhx.queue.TopicPartitionInfo;
@@ -40,19 +41,19 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
 
     @Override
     public void subscribe() {
-        log.info("enqueue topic subscribe {} ", topic);
+
         if (stopped) {
-            log.error("trying subscribe, but consumer stopped for topic {}", topic);
+            log.error("consumer 线程已停止 topic {}", topic);
             return;
         }
-        subscribeQueue.add(Collections.singleton(new TopicPartitionInfo(topic, null)));
+        subscribeQueue.add(Collections.singleton(new TopicPartitionInfo(topic, null,"")));
     }
 
     @Override
     public void subscribe(Set<TopicPartitionInfo> partitions) {
-        log.info("enqueue topics subscribe {} ", partitions);
+        log.info("订阅的topics {} ", JSON.toJSONString(partitions));
         if (stopped) {
-            log.error("trying subscribe, but consumer stopped for topic {}", topic);
+            log.error("订阅服务已停止，topic {}", topic);
             return;
         }
         subscribeQueue.add(partitions);
@@ -60,7 +61,6 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
 
     @Override
     public List<T> poll(long durationInMillis) {
-
         List<R> records;
         long startNanos = System.nanoTime();
         if (stopped) {
@@ -73,7 +73,6 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
         if (consumerLock.isLocked()) {
             log.error("poll. consumerLock is locked. will wait with no timeout. it looks like a race conditions or deadlock", new RuntimeException("stacktrace"));
         }
-
         consumerLock.lock();
         try {
             //更新订阅的主题
@@ -169,5 +168,4 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
     abstract protected void doUnsubscribe();
 
 
-
 }

common-lib/src/main/java/com/mortals/xhx/queue/rabbitmq/TbRabbitMqConsumerTemplate.java

@@ -56,9 +56,8 @@ public class TbRabbitMqConsumerTemplate<T extends TbQueueMsg> extends AbstractTb
                         GetResponse getResponse = channel.basicGet(queue, true);
                         return getResponse;
                     } catch (IOException e) {
-                        // log.error("Failed to get messages from queue: [{}]" , queue);
+                        log.error("Failed to get messages from queue: [{}]" , queue,e);
                         return null;
-                        //  throw new RuntimeException("Failed to get messages from queue." , e);
                     }
                 }).filter(Objects::nonNull).collect(Collectors.toList());
         if (result.size() > 0) {
@@ -73,7 +72,6 @@ public class TbRabbitMqConsumerTemplate<T extends TbQueueMsg> extends AbstractTb
         queues = partitions.stream()
                 .map(TopicPartitionInfo::getFullTopicName)
                 .collect(Collectors.toSet());
-        //queues.forEach(admin::createTopicIfNotExists);
     }
 
     @Override

common-lib/src/main/java/com/mortals/xhx/queue/rabbitmq/TbRabbitMqProducerTemplate.java

@@ -6,6 +6,7 @@ import com.google.common.util.concurrent.MoreExecutors;
 import com.google.gson.Gson;
 import com.mortals.xhx.queue.*;
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.BuiltinExchangeType;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import lombok.extern.slf4j.Slf4j;
@@ -16,15 +17,37 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 
+/**
+ * rabbmit 消息生产模板
+ *
+ * @author: zxfei
+ * @date: 2022/4/25 16:10
+ */
 @Slf4j
 public class TbRabbitMqProducerTemplate<T extends TbQueueMsg> implements TbQueueProducer<T> {
+    /**
+     * defaultTopic
+     */
     private String defaultTopic;
+    /**
+     * rabbmit设置
+     */
     private TbRabbitMqSettings rabbitMqSettings;
+    /**
+     * 线程执行器
+     */
     private ListeningExecutorService producerExecutor;
+    /**
+     * 通道
+     */
     private Channel channel;
+    /**
+     * 连接器
+     */
     private Connection connection;
-    private final Gson gson = new Gson();
-
+    /**
+     * topic组
+     */
     private final Set<TopicPartitionInfo> topics = ConcurrentHashMap.newKeySet();
 
     public TbRabbitMqProducerTemplate(TbRabbitMqSettings rabbitMqSettings, String defaultTopic) {
@@ -35,8 +58,7 @@ public class TbRabbitMqProducerTemplate<T extends TbQueueMsg> implements TbQueue
             connection = rabbitMqSettings.getConnectionFactory().newConnection();
             channel = connection.createChannel();
         } catch (IOException | TimeoutException e) {
-            log.error("Failed to create connection.", e);
-            // throw new RuntimeException("Failed to create connection." , e);
+            log.error("rabbmit创建连接失败！", e);
         }
 
     }
@@ -55,23 +77,19 @@ public class TbRabbitMqProducerTemplate<T extends TbQueueMsg> implements TbQueue
     public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) {
         Boolean topicIfNotExist = createTopicIfNotExist(tpi);
         AMQP.BasicProperties properties = new AMQP.BasicProperties();
-//                .builder()
-//                .contentType("application/json")
-//                .deliveryMode(2) // 消息是否持久化，1未持久，2持久
-//                .contentEncoding("UTF-8") // 编码方式
-//                .expiration("100000") // 过期时间单位毫秒
-//                .build();
         try {
             if (!topicIfNotExist) {
                 channel.queueDeclare(tpi.getTopic(), true, false, false, null);
             }
-            //channel.basicPublish(rabbitMqSettings.getExchangeName(), tpi.getFullTopicName(), properties, gson.toJson(new DefaultTbQueueMsg(msg)).getBytes());
-            channel.basicPublish(rabbitMqSettings.getExchangeName(), tpi.getFullTopicName(), properties, JSON.toJSONString(new DefaultTbQueueMsg(msg)).getBytes());
+            if (!innerExists(tpi.getExchangeName(), channel)) {
+                channel = connection.createChannel();
+                channel.exchangeDeclare(tpi.getExchangeName(), BuiltinExchangeType.DIRECT);
+            }
+            channel.basicPublish(tpi.getExchangeName(), tpi.getFullTopicName(), properties, JSON.toJSONString(new DefaultTbQueueMsg(msg)).getBytes());
             if (callback != null) {
                 callback.onSuccess(null);
             }
-        } catch (
-                IOException e) {
+        } catch (IOException e) {
             log.error("Failed publish message: [{}].", msg, e);
             if (callback != null) {
                 callback.onFailure(e);
@@ -108,4 +126,15 @@ public class TbRabbitMqProducerTemplate<T extends TbQueueMsg> implements TbQueue
         topics.add(tpi);
         return false;
     }
+
+
+    private boolean innerExists(String exchangeName, Channel outerChannel) {
+        boolean result = true;
+        try {
+            outerChannel.exchangeDeclarePassive(exchangeName);
+        } catch (IOException e) {
+            result = false;
+        }
+        return result;
+    }
 }

device-manager-ui/admin/src/components/Header.vue

@@ -98,7 +98,7 @@ export default {
     });
 
        let _this = this;
-    const getsocketData = (e) => {
+    const getsocketData = (e) => {i
       // 创建接收消息函数
       const data = e && e.detail.data;
 

device-manager/src/main/java/com/mortals/xhx/base/framework/config/P6spySqlFormatConfig.java

@@ -16,6 +16,7 @@ public class P6spySqlFormatConfig implements MessageFormattingStrategy {
 
     @Override
     public String formatMessage(int connectionId, String now, long elapsed, String category, String prepared, String sql, String url) {
+
         return StringUtils.isNotBlank(sql) ? DateUtils.getCurrStrDateTime()
                 + " | 耗时 " + elapsed + " ms | SQL:" + StringUtils.LF + sql.replaceAll("[\\s]+", StringUtils.SPACE) + ";" : "";
     }

device-manager/src/main/java/com/mortals/xhx/busiz/web/DeviceApiController.java

@@ -148,7 +148,7 @@ public class DeviceApiController {
      * @return
      */
     @PostMapping("upload")
-    @DeviceAuth
+    //@DeviceAuth
     public String upload(@RequestBody UploadDeviceReq req) {
         log.info("【设备数据上报】【请求体】-->  " + JSONObject.toJSONString(req));
         ApiResp<String> rsp = new ApiResp<>();
@@ -159,7 +159,19 @@ public class DeviceApiController {
             DeviceEntity deviceEntity = deviceService.getExtCache(req.getDeviceNum());
             if (!ObjectUtils.isEmpty(deviceEntity)) {
                 //将上报信息转发到mq中
-                TopicPartitionInfo info = TopicPartitionInfo.builder().topic(Constant.UPLOAD_TOPIC + deviceEntity.getDeviceMac()).build();
+
+                PlatformEntity platformEntity = platformService.get(deviceEntity.getPlatformId());
+                if (ObjectUtils.isEmpty(platformEntity)) {
+                    throw new AppException("当前设备未配置所属系统平台，请在后台配置后再激活！");
+                }
+                // authInfo.setHost(platformEntity.getPlatformSn());
+                ProductEntity productEntity = productService.get(deviceEntity.getProductId());
+                if (ObjectUtils.isEmpty(productEntity)) {
+                    throw new AppException("当前设备未配置所属产品，请在后台配置后再激活！");
+                }
+                String exchangeName = platformEntity.getPlatformSn() + Constant.EXCHANGE_SPLIT + productEntity.getProductCode();
+
+                TopicPartitionInfo info = TopicPartitionInfo.builder().exchangeName(exchangeName).topic(Constant.UPLOAD_TOPIC + deviceEntity.getDeviceMac()).build();
                 TbQueueMsgHeaders header = new DefaultTbQueueMsgHeaders();
                 header.put(MessageHeader.MESSAGETYPE, Constant.MESSAGETYPE_HEARTBEAT);
                 ApiResp<String> sendDeviceMessageResp = deviceService.sendDeviceMessage(deviceEntity, info, header, JSON.toJSONString(req), null);

device-manager/src/main/java/com/mortals/xhx/daemon/applicationservice/DeviceMsgComsumerStartedService.java

@@ -59,7 +59,7 @@ public class DeviceMsgComsumerStartedService implements IApplicationStartedServi
             //订阅所有已几快活设备
             Set<TopicPartitionInfo> topicPartitionInfoSet = deviceService.find(new DeviceQuery().active(ActiveEnum.已激活.getValue()).status(StatusEnum.启用.getValue())).stream()
                     .map(item ->
-                            new TopicPartitionInfo(Constant.UPLOAD_TOPIC + item.getDeviceMac(), null)
+                            new TopicPartitionInfo(Constant.UPLOAD_TOPIC + item.getDeviceMac(), null,null)
                     ).collect(Collectors.toSet());
             mainConsumer.subscribe(topicPartitionInfoSet);
             log.info("消费线程订阅设备上报消息成功！：" + JSON.toJSONString(topicPartitionInfoSet));

device-manager/src/main/resources/bootstrap.yml

@@ -105,7 +105,7 @@ token:
   # 令牌自定义标识
   header: Authorization
   # 令牌密钥
-  secret: abcd1234
+  secret: 026db82420614469897fcc2dc1b4ce38
   # 令牌有效期（默认60分钟）
   expireTime: 60
   # 令牌前缀

device-manager/src/main/resources/spy.properties

@@ -16,4 +16,7 @@ driverlist=com.mysql.cj.jdbc.Driver
 # 是否开启慢SQL记录
 outagedetection=true
 # 慢SQL记录标准 秒
-outagedetectioninterval=2
\ No newline at end of file
+outagedetectioninterval=2
+
+filter=true
+exclude=mortals_xhx_task
\ No newline at end of file

device-manager/src/test/java/com/mortals/httpclient/device/DeviceController.http

@@ -54,7 +54,7 @@ POST {{baseUrl}}/api/active
 Content-Type: application/json
 
 {
-  "deviceNum":"b12345678",
+  "deviceNum":"AB:DD:DF:FD:AD:FA:DA:SS",
   "deviceMac":"AB:DD:DF:FD:AD:FA:DA:SS",
   "action":"active"
 }
@@ -71,7 +71,7 @@ Content-Type: application/json
 Authorization: Bearer  {{authToken}}
 
 {
-  "deviceNum":"a12345678",
+  "deviceNum":"AB:DD:DF:FD:AD:FA:DA:SS",
   "action":"upload"
 }