tda

hx-admin/src/main/resources/application-test.yml

@@ -44,3 +44,7 @@ spring:
                 max-active: 8
                 # #连接池最大阻塞等待时间（使用负值表示没有限制）
                 max-wait: -1ms
+
+server:
+    servlet:
+        context-path: /test-api

hx-iot/src/main/java/com/fjhx/iot/controller/tda/TdaDeviceController.java

@@ -1,5 +1,7 @@
 package com.fjhx.iot.controller.tda;
 
+import com.baomidou.dynamic.datasource.annotation.DS;
+import com.fjhx.common.constant.SourceConstant;
 import org.springframework.web.bind.annotation.*;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.fjhx.iot.entity.tda.vo.TdaDeviceVo;
@@ -18,6 +20,7 @@ import org.springframework.beans.factory.annotation.Autowired;
  * @author zlj
  * @since 2023-03-21
  */
+@DS(SourceConstant.IOT)
 @RestController
 @RequestMapping("/tdaDevice")
 public class TdaDeviceController {

hx-iot/src/main/java/com/fjhx/iot/controller/tda/TdaProductController.java

@@ -1,6 +1,8 @@
 package com.fjhx.iot.controller.tda;
 
+import com.baomidou.dynamic.datasource.annotation.DS;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.fjhx.common.constant.SourceConstant;
 import com.fjhx.iot.entity.tda.dto.TdaProductDto;
 import com.fjhx.iot.entity.tda.dto.TdaProductSelectDto;
 import com.fjhx.iot.entity.tda.vo.TdaProductVo;
@@ -21,6 +23,7 @@ import org.springframework.web.bind.annotation.RestController;
  * @author zlj
  * @since 2023-03-21
  */
+@DS(SourceConstant.IOT)
 @RestController
 @RequestMapping("/tdaProduct")
 public class TdaProductController {

hx-iot/src/main/java/com/fjhx/iot/listener/DtaListener.java

@@ -67,6 +67,8 @@ public class DtaListener {
     }
 
     public static MessageConsumer getMessageConsumer(TdaConfig tdaConfig) {
+
+        MessageConsumer consumer = null;
         try {
             AmqpClientOptions options = AmqpClientOptions.builder()
                     .host(tdaConfig.getEndPoint())
@@ -77,11 +79,18 @@ public class DtaListener {
                     .build();
             AmqpClient amqpClient = new AmqpClient(options);
             amqpClient.initialize();
-            MessageConsumer consumer = amqpClient.newConsumer(AmqpConstants.DEFAULT_QUEUE);
+            consumer = amqpClient.newConsumer(AmqpConstants.DEFAULT_QUEUE);
             // 此处进行消息处理。如果处理比较耗时，最好进行开启新的线程处理，否则可能造成心跳超时链接断开。
             consumer.setMessageListener(DtaListener::handleData);
             return consumer;
         } catch (Exception e) {
+            try {
+                if (consumer != null) {
+                    consumer.close();
+                }
+            } catch (JMSException ex) {
+                log.error("关闭失败", ex);
+            }
             log.error("链接失败，tdaConfig:{}", tdaConfig.getId(), e);
             throw new ServiceException("连接iot失败");
         }