add

mqtt信息发送

nb-admin/src/main/resources/application-dev.yml

@@ -183,18 +183,6 @@ oss:
     accessSecret: "R7hOvMfiHb0PYroDqUDXAYgB9htQss"
     endpoint: "oss-cn-hangzhou.aliyuncs.com"
     bucketName: "tuoren-nb-dev"
-mqtt:
-  client:
-    id: nb-netpump-client-dev
-  broker:
-    url: tcp://192.168.100.115:1883
-    username: hospital
-    password: 1qaz!QAZ
-  connect-options:
-    clean-session: true
-    connection-timeout: 30
-    keep-alive-interval: 60
-    automatic-reconnect: true
 notify:
   wechat:
     url: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=c3e093fe-5125-47d5-a171-0f4be2f61a78

nb-core/src/main/java/com/nb/core/config/MqttConfig.java

@@ -52,6 +52,8 @@ public class MqttConfig {
         }
         // 设置自动重连
         options.setAutomaticReconnect(mqttProperties.getConnectOptions().isAutomaticReconnect());
+        // 设置重连相关参数
+        options.setMaxReconnectDelay(mqttProperties.getConnectOptions().getMaxReconnectDelay());
         factory.setConnectionOptions(options);
         return factory;
     }

nb-core/src/main/java/com/nb/core/config/MqttProperties.java

@@ -48,7 +48,7 @@ public class MqttProperties {
         /**
          * 代理URL
          */
-        private String url = "tcp://192.168.100.115:1883";
+        private String url = "tcp://iot.tuoren.com:1883";
         
         /**
          * 用户名
@@ -106,6 +106,16 @@ public class MqttProperties {
          */
         private boolean automaticReconnect = true;
         
+        /**
+         * 重连间隔（毫秒）
+         */
+        private int reconnectDelay = 5000;
+        
+        /**
+         * 最大重连延迟（毫秒）
+         */
+        private int maxReconnectDelay = 60000;
+        
         public boolean isCleanSession() {
             return cleanSession;
         }
@@ -137,6 +147,22 @@ public class MqttProperties {
         public void setAutomaticReconnect(boolean automaticReconnect) {
             this.automaticReconnect = automaticReconnect;
         }
+        
+        public int getReconnectDelay() {
+            return reconnectDelay;
+        }
+        
+        public void setReconnectDelay(int reconnectDelay) {
+            this.reconnectDelay = reconnectDelay;
+        }
+        
+        public int getMaxReconnectDelay() {
+            return maxReconnectDelay;
+        }
+        
+        public void setMaxReconnectDelay(int maxReconnectDelay) {
+            this.maxReconnectDelay = maxReconnectDelay;
+        }
     }
     
     // Getters and Setters

nb-core/src/main/java/com/nb/core/utils/MqttClientUtil.java

@@ -52,7 +52,7 @@ public class MqttClientUtil implements ApplicationContextAware {
         try {
             mqttClientFactory = applicationContext.getBean(MqttPahoClientFactory.class);
         } catch (Exception e) {
-
+            // 忽略异常，如果获取不到工厂则无法进行MQTT操作
         }
     }
 
@@ -96,14 +96,14 @@ public class MqttClientUtil implements ApplicationContextAware {
      */
     public void publish(String hospitalCode, String topic, Object payload, int qos, boolean retained) {
         try {
-            // 创建消息实体
-            MqttMessage message = new MqttMessage(topic, JSONUtil.toJsonStr( payload));
-            message.setClientId(CLIENT_ID);
-            
             if (mqttClientFactory == null) {
                 return;
             }
             
+            // 创建消息实体
+            MqttMessage message = new MqttMessage(topic, JSONUtil.toJsonStr(payload));
+            message.setClientId(CLIENT_ID);
+            
             MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(CLIENT_ID, mqttClientFactory);
             messageHandler.setDefaultTopic("hospitalInfo/"+hospitalCode);
             messageHandler.setDefaultQos(qos);
@@ -125,14 +125,14 @@ public class MqttClientUtil implements ApplicationContextAware {
      * @param topic 主题
      * @param messageHandler 消息处理器
      */
-    public void subscribe( String topic, MessageHandler messageHandler) {
+    public void subscribe(String topic, MessageHandler messageHandler) {
         try {
             if (mqttClientFactory == null) {
                 return;
             }
             // 如果已经订阅了该主题，则先取消订阅
             if (subscribers.containsKey(topic)) {
-                unsubscribe( topic);
+                unsubscribe(topic);
             }
             
             MqttPahoMessageDrivenChannelAdapter adapter = 
@@ -173,12 +173,15 @@ public class MqttClientUtil implements ApplicationContextAware {
      * 取消订阅
      * @param topic 主题
      */
-    public void unsubscribe( String topic) {
-        String key = topic;
-        MqttPahoMessageDrivenChannelAdapter adapter = subscribers.get(key);
+    public void unsubscribe(String topic) {
+        MqttPahoMessageDrivenChannelAdapter adapter = subscribers.get(topic);
         if (adapter != null) {
-            adapter.stop();
-            subscribers.remove(key);
+            try {
+                adapter.stop();
+            } catch (Exception e) {
+                // 忽略停止异常
+            }
+            subscribers.remove(topic);
         }
     }
 

nb-service/web-service/src/main/java/com/nb/web/service/bus/listener/BusEvaluationMessageListener.java

@@ -8,9 +8,7 @@ import com.nb.core.entity.MqttMessage;
 import com.nb.core.enums.MqttTopicEnum;
 import com.nb.core.handler.AbstractMqttMessageHandler;
 import com.nb.web.api.entity.BusEvaluationEntity;
-import com.nb.web.api.entity.BusInfusionHistoryEntity;
 import com.nb.web.service.bus.entity.BusPatientEntity;
-import com.nb.web.service.bus.mapper.BusInfusionHistoryMapper;
 import com.nb.web.service.bus.mapper.BusPatientMapper;
 import com.nb.web.service.bus.service.LocalBusEvaluationService;
 import lombok.extern.slf4j.Slf4j;
@@ -33,9 +31,6 @@ public class BusEvaluationMessageListener extends AbstractMqttMessageHandler {
 
     @Autowired
     private BusPatientMapper patientMapper;
-
-    @Autowired
-    private BusInfusionHistoryMapper infusionHistoryMapper;
     /**
      * 定义支持的主题模式
      * @return 支持的主题模式
@@ -72,11 +67,6 @@ public class BusEvaluationMessageListener extends AbstractMqttMessageHandler {
             evaluation.setInfusionId(patient.getInfusionId());
             evaluation.setClinicId(patient.getClinicId());
             evaluation.setTenantId(patient.getTenantId());
-            BusInfusionHistoryEntity infusionHistory = infusionHistoryMapper.selectById(patient.getInfusionId());
-            if(ObjectUtil.isNotNull(infusionHistory)){
-                evaluation.setDeviceId(infusionHistory.getDeviceId());
-            }
-            evaluation.setId(null);
             // 保存到数据库
             evaluationService.save(evaluation);
             log.info("评价信息处理完成，ID: {}", evaluation.getId());