添加巴中经开区windows shell

device-manager/src/main/java/com/mortals/xhx/base/framework/exception/ExceptionHandle.java

 package com.mortals.xhx.base.framework.exception;
 
 import cn.hutool.core.util.StrUtil;
-import cn.hutool.http.HttpStatus;
 import com.alibaba.fastjson.JSONObject;
 import com.mortals.framework.exception.AppException;
 import lombok.extern.slf4j.Slf4j;
@@ -46,7 +45,7 @@ public class ExceptionHandle {
         }
         if (e instanceof HttpMessageNotReadableException) {
             log.error("[system error]", e);
-            response.setStatus(HttpStatus.HTTP_BAD_REQUEST);
+         //   response.setStatus(HttpStatus.HTTP_BAD_REQUEST);
             ret.put(KEY_RESULT_MSG, "参数错误，"+ StrUtil.subBefore(e.getMessage(), ";", false));
         } else {
             log.error("[system error]", e);

device-manager/src/main/java/com/mortals/xhx/base/framework/listener/DirectDynamicListener.java

@@ -13,7 +13,6 @@ import com.mortals.xhx.common.code.LogTypeEnum;
 import com.mortals.xhx.common.key.Constant;
 import com.mortals.xhx.common.key.RedisKey;
 import com.mortals.xhx.common.pdu.DefaultQueueMsg;
-import com.mortals.xhx.common.utils.SendTaskThreadPool;
 import com.mortals.xhx.module.device.model.DeviceEntity;
 import com.mortals.xhx.module.device.model.DeviceLogEntity;
 import com.mortals.xhx.module.device.service.DeviceLogService;
@@ -53,8 +52,6 @@ public class DirectDynamicListener implements MessageListener {
     @Autowired
     private PlatformService platformService;
     @Autowired
-    private SendTaskThreadPool sendTaskThreadPool;
-    @Autowired
     private ICacheService cacheService;
     @Autowired
     private IMessageProduceService messageProducer;
@@ -84,16 +81,7 @@ public class DirectDynamicListener implements MessageListener {
                     condition.setId(deviceEntity.getId());
                     condition.setOnlineTime(new Date());
                     condition.setDeviceStatus(DeviceStatusEnum.在线.getValue());
-//                    DeviceEntity entity = new DeviceEntity();
-//                    entity.setOnlineTime(new Date());
-//                    entity.setDeviceStatus(DeviceStatusEnum.在线.getValue());
-//                    DeviceEntity condition = new DeviceEntity();
-//                    condition.setId(deviceEntity.getId());
-                    //int update = deviceService.getDeviceDao().update(entity, condition);
-                    // deviceService.update(deviceEntity);
-                     //deviceService.putCache(deviceEntity.getId().toString(),deviceEntity);
-
-                     cacheService.lpush(RedisKey.KEY_DEVICE_UPDATE_QUEUE,condition);
+                    cacheService.lpush(RedisKey.KEY_DEVICE_UPDATE_QUEUE, condition);
 
                     PlatformEntity platformEntity = platformService.getCache(deviceEntity.getPlatformId().toString());
                     ProductEntity productEntity = productService.get(deviceEntity.getProductId());
@@ -114,7 +102,7 @@ public class DirectDynamicListener implements MessageListener {
                         bizLogPdu.setLogTime(new Date());
                         messageProducer.syncBizSend(bizLogPdu);
                     }
-                   // log.info("id:{},deviceCode:{} deviceStatus:{}==>上线  ", deviceEntity.getId(), deviceEntity.getDeviceCode(), deviceEntity.getDeviceStatus());
+                    // log.info("id:{},deviceCode:{} deviceStatus:{}==>上线  ", deviceEntity.getId(), deviceEntity.getDeviceCode(), deviceEntity.getDeviceStatus());
                 }
                 if (!Constant.MESSAGETYPE_HEARTBEAT.equalsIgnoreCase(messageType)) {
                     DeviceLogEntity deviceLogEntity = new DeviceLogEntity();
@@ -131,25 +119,9 @@ public class DirectDynamicListener implements MessageListener {
                     deviceLogEntity.setCreateTime(new Date());
                     deviceLogService.save(deviceLogEntity);
                 }
-            /*    //获取exchange,
-                PlatformEntity platformEntity = platformService.getCache(deviceEntity.getPlatformId().toString());
-
-                String phpInUrl = GlobalSysInfo.getParamValue(PARAM_SERVER_PHP_IN_HTTP_URL, "http://172.15.28.116:8090");
-                if (!Constant.MESSAGETYPE_HEARTBEAT.equalsIgnoreCase(messageType)
-                        && !ObjectUtils.isEmpty(platformEntity)
-                        && platformEntity.getSendSwitch() == YesNoEnum.YES.getValue()) {
-                    if (platformEntity.getSendMsgType() == SendMsgTypeEnum.http.getValue()) {
-                        SendTask sendTask = new SendTask(phpInUrl, queueMsg.getData());
-                        sendTaskThreadPool.execute(sendTask);
-                    }
-                }*/
             }
         } catch (Exception e) {
             log.error("消费消息异常", e);
         }
-
-//        log.info("接收到：" + queue + "，消息内容为：" + new String(body));
-//        //channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
-//        log.info(queue + "队列消息消费成功");
     }
 }

device-manager/src/main/java/com/mortals/xhx/base/framework/listener/ListenerContainerConsumerFailedEventListener.java

+package com.mortals.xhx.base.framework.listener;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.listener.ListenerContainerConsumerFailedEvent;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.context.ApplicationListener;
+import org.springframework.stereotype.Component;
+import org.springframework.util.Assert;
+
+import java.util.Arrays;
+
+/**
+ * MQ消费者失败事件监听器
+ * @author: zxfei
+ * @date: 2024/10/18 9:50
+ */
+@Slf4j
+@Component
+public class ListenerContainerConsumerFailedEventListener implements ApplicationListener<ListenerContainerConsumerFailedEvent> {
+
+    @Override
+    public void onApplicationEvent(ListenerContainerConsumerFailedEvent event) {
+        log.error("消费者失败事件发生：{}", event);
+
+        if (event.isFatal()) {
+            log.error(String.format("Stopping container from aborted consumer. Reason::%s.",
+                    event.getReason()), event.getThrowable());
+            SimpleMessageListenerContainer container = (SimpleMessageListenerContainer) event.getSource();
+            String queueNames = Arrays.toString(container.getQueueNames());
+            // 重启
+            try {
+                restart(container);
+                log.info("重启队列%s的监听成功！", queueNames);
+            } catch (Exception e) {
+                log.error(String.format("重启队列%s的监听失败！", queueNames), e);
+            }
+
+            // TODO 告警，包含队列信息，监听断开原因，断开时异常信息，重启是否成功等...
+
+        }
+    }
+
+    /**
+     * 重启监听
+     * @param container
+     * @return
+     */
+    private void restart(SimpleMessageListenerContainer container) {
+        // 暂停30s
+        try {
+            Thread.sleep(30000);
+        } catch (Exception e) {
+            log.error(e.getMessage());
+        }
+
+        Assert.state(!container.isRunning(), String.format("监听容器%s正在运行！", container));
+        container.start();
+    }
+
+}
\ No newline at end of file

device-manager/src/main/java/com/mortals/xhx/base/framework/listener/RabbitLoggingErrorHandler.java

 package com.mortals.xhx.base.framework.listener;
 
 import lombok.extern.slf4j.Slf4j;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
 import org.springframework.stereotype.Component;
 import org.springframework.util.ErrorHandler;
@@ -17,8 +15,7 @@ public class RabbitLoggingErrorHandler implements ErrorHandler {
 
     @Override
     public void handleError(Throwable t) {
-
-        log.error("[handleError][发生异常]]", t);
+        log.error("[handleError][发生异常]]  {}", t.getMessage());
     }
 
 }