mq修改

themis-business/src/main/java/com/qmth/themis/business/constant/SystemConstant.java

@@ -128,8 +128,8 @@ public class SystemConstant {
      */
     public static final String WEBSOCKET_OE_ONLINE_COUNT = "websocket:oe:online:count";
     public static final String GET = "get";
-//    public static final long WEBSOCKET_MAX_TIME_OUT = 3 * 60 * 1000;
-    public static final long WEBSOCKET_MAX_TIME_OUT = 1 * 60 * 1000;
+    public static final long WEBSOCKET_MAX_TIME_OUT = 3 * 60 * 1000;
+//    public static final long WEBSOCKET_MAX_TIME_OUT = 1 * 60 * 1000;
     //        public static final long WEBSOCKET_MAX_TIME_OUT = 10 * 1000;
     public static final String ACK_MESSAGE = "ackMessage";
     /**

themis-exam/src/main/java/com/qmth/themis/exam/websocket/WebSocketOeServer.java

@@ -166,12 +166,12 @@ public class WebSocketOeServer implements Concurrently {
                 log.info("超时退出");
                 //发送延时mq消息start
                 MqDtoService mqDtoService = SpringContextHolder.getBean(MqDtoService.class);
-//                String level = "2m";
-                String level = "30s";
+                String level = "2m";
+//                String level = "30s";
                 Integer time = SystemConstant.mqDelayLevel.get(level);
                 LocalDateTime dt = LocalDateTime.now();
-//                dt = dt.plusMinutes(Long.parseLong(level.replace("m", "")));
-                dt = dt.plusMinutes(Long.parseLong(level.replace("s", "")));
+                dt = dt.plusMinutes(Long.parseLong(level.replace("m", "")));
+//                dt = dt.plusSeconds(Long.parseLong(level.replace("s", "")));
                 tranMap.put("timeOut", time);
                 tranMap.put("mqExecTime", dt.toInstant(ZoneOffset.of("+8")).toEpochMilli());
                 MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.unNormal.name(), SystemOperationEnum.OE_NET_UN_NORMAL, MqEnum.WEBSOCKET_UN_NORMAL_LOG, String.valueOf(this.recordId), this.tranMap, this.sessionId);

themis-mq/src/main/java/com/qmth/themis/mq/service/impl/MqDtoServiceImpl.java

@@ -156,6 +156,7 @@ public class MqDtoServiceImpl implements MqDtoService {
     }
 
     /**
+     * 设置mq buffer type
      *
      * @param mqDto
      */

themis-mq/src/main/java/com/qmth/themis/mq/service/impl/MqLogicServiceImpl.java

@@ -223,7 +223,7 @@ public class MqLogicServiceImpl implements MqLogicService {
         map.put(SystemConstant.MQDTO_OBJ, JacksonUtil.parseJson(mqDto));
         tmRocketMessage.setProp(JacksonUtil.parseJson(map));
         tmRocketMessageService.saveOrUpdate(tmRocketMessage);
-//        redisUtil.delete(key, mqDto.getId());
+        redisUtil.delete(key, mqDto.getId());
     }
 
     /**

themis-mq/src/main/java/com/qmth/themis/mq/templete/impl/WebsocketUnNormalConcurrentlyImpl.java

@@ -49,7 +49,6 @@ public class WebsocketUnNormalConcurrentlyImpl implements Concurrently {
                 log.info(":{}-:{} websocket unnormal Consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
                 int reconsumeTime = messageExt.getReconsumeTimes();
                 if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    //超过最大重试次数，保存到数据库，后续可以发短信通知系统管理人员
                     mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_DELAY_TOPIC_BUFFER_LIST);
                 } else {
                     if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(SystemConstant.MQ_DELAY_TOPIC_BUFFER_LIST) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {