Merge branch 'dev'

1

themis-business/src/main/java/com/qmth/themis/business/enums/MqExecTypeEnum.java

@@ -0,0 +1,72 @@
+package com.qmth.themis.business.enums;
+
+import java.util.Objects;
+
+/**
+ * @Description: mq exec类型 enum
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/9/23
+ */
+public enum MqExecTypeEnum {
+
+    EXEC_MQ_SESSION_LOGIC("session逻辑", "execMqSessionLogic"),
+
+    EXEC_MQ_LOG_LOGIC("log逻辑", "execMqLogLogic"),
+
+    EXEC_MQ_USER_LOG_LOGIC("userLog逻辑", "execMqUserLogLogic"),
+
+    EXEC_MQ_TASK_LOGIC("task逻辑", "execMqTaskLogic"),
+
+    EXEC_MQ_LIVENESS_VERIFY_LOGIC("活体检测逻辑", "execMqLivenessVerifySaveLogic"),
+
+    EXEC_MQ_FACE_VERIFY_LOGIC("人脸检测逻辑", "execMqFaceVerifySaveLogic"),
+
+    EXEC_MQ_EXAM_STUDENT_UPDATE_LOGIC("考生更新逻辑", "execMqExamStudentUpdateLogic"),
+
+    EXEC_MQ_RECORD_UPDATE_LOGIC("考试记录更新逻辑", "execMqRecordUpdateLogic"),
+
+    EXEC_MQ_RECORD_PERSISTED_LOGIC("考试记录持久化逻辑", "execMqRecordPersistedLogic"),
+
+    EXEC_MQ_RECORD_INIT_LOGIC("考试记录初始化逻辑", "execMqRecordInitLogic"),
+
+    EXEC_MQ_EXAM_BREAK_LOGIC("考试断点逻辑", "execMqExamBreakLogic"),
+
+    EXEC_MQ_CALCULATE_SCORE_LOGIC("重新算分逻辑", "execMqCalculateScoreLogic"),
+
+    EXEC_MQ_CALCULATE_OBJECTIVE_SCORE_LOGIC("计算客观分逻辑", "execMqCalculateObjectiveScoreLogic"),
+
+    EXEC_MQ_WEBSOCKET_UN_NORMAL_LOGIC("websocket非正常退出，延时消息逻辑", "execMqWebsocketUnNormalLogic");
+
+    private String code;
+    private String desc;
+
+    private MqExecTypeEnum(String code, String desc) {
+        this.code = code;
+        this.desc = desc;
+    }
+
+    public String getCode() {
+        return code;
+    }
+
+    public String getDesc() {
+        return desc;
+    }
+
+    /**
+     * 状态转换 toName
+     *
+     * @param value
+     * @return
+     */
+    public static String convertToName(String value) {
+        for (MqExecTypeEnum e : MqExecTypeEnum.values()) {
+            if (Objects.equals(value.trim(), e.getDesc())) {
+                return e.name();
+            }
+        }
+        return null;
+    }
+}

themis-mq/src/main/java/com/qmth/themis/mq/service/MqLogicService.java

@@ -1,6 +1,12 @@
 package com.qmth.themis.mq.service;
 
 import com.qmth.themis.business.dto.MqDto;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.common.message.MessageExt;
+
+import java.util.List;
 
 /**
  * @Description: mq执行逻辑
@@ -81,7 +87,7 @@ public interface MqLogicService {
      * @param mqDto
      * @param key
      */
-    public void execMqExamRecordPersistedLogic(MqDto mqDto, String key);
+    public void execMqRecordPersistedLogic(MqDto mqDto, String key);
 
     /**
      * 考试记录数据更新
@@ -127,4 +133,24 @@ public interface MqLogicService {
      * @param key
      */
     void execMqExamBreakLogic(MqDto mqDto, String key);
+
+    /**
+     * 普通消息
+     *
+     * @param msgs
+     * @param consumeConcurrentlyContext
+     * @param mqExecTypeEnum
+     * @return
+     */
+    ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext, MqExecTypeEnum mqExecTypeEnum);
+
+    /**
+     * 延时消息
+     *
+     * @param msgs
+     * @param consumeConcurrentlyContext
+     * @param mqExecTypeEnum
+     * @return
+     */
+    ConsumeConcurrentlyStatus consumeMessageDelay(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext, MqExecTypeEnum mqExecTypeEnum);
 }

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

@@ -28,6 +28,9 @@ import com.qmth.themis.business.util.RedisUtil;
 import com.qmth.themis.common.contanst.Constants;
 import com.qmth.themis.common.util.SimpleBeanUtil;
 import com.qmth.themis.mq.service.MqLogicService;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.common.message.MessageExt;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
@@ -35,6 +38,7 @@ import org.springframework.transaction.annotation.Transactional;
 
 import javax.annotation.Resource;
 import java.io.IOException;
+import java.lang.reflect.Method;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
@@ -410,7 +414,7 @@ public class MqLogicServiceImpl implements MqLogicService {
      */
     @Override
     @Transactional
-    public void execMqExamRecordPersistedLogic(MqDto mqDto, String key) {
+    public void execMqRecordPersistedLogic(MqDto mqDto, String key) {
         Gson gson = new Gson();
         Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
         Long recordId = (Long) param.get("recordId");
@@ -632,4 +636,91 @@ public class MqLogicServiceImpl implements MqLogicService {
         tmRocketMessageService.saveOrUpdate(tmRocketMessage);
         redisUtil.delete(key, mqDto.getId());
     }
+
+    /**
+     * 普通消息
+     *
+     * @param msgs
+     * @param consumeConcurrentlyContext
+     * @param mqExecTypeEnum
+     * @return
+     */
+    @Override
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext, MqExecTypeEnum mqExecTypeEnum) {
+        MqDto mqDto = null;
+        try {
+            long threadId = Thread.currentThread().getId();
+            String threadName = Thread.currentThread().getName();
+            for (MessageExt messageExt : msgs) {
+                log.info(":{}-:{} consumer 重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+                log.info(":{}-:{} consumer 接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
+                int reconsumeTime = messageExt.getReconsumeTimes();
+                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
+                    this.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
+                } else {
+                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
+                        Method method = this.getClass().getDeclaredMethod(mqExecTypeEnum.getDesc(), MqDto.class, String.class);
+                        log.info(":{}-:{} 准备执行mq exec:{}逻辑", threadId, threadName, method.getName());
+                        method.invoke(this, mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
+                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                    } else {
+                        log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
+                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                    }
+                }
+            }
+        } catch (Exception e) {
+            log.error("mq普通消息监听，消息消费出错", e);
+            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+        } finally {
+            if (Objects.nonNull(mqDto)) {
+                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
+            }
+        }
+        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+    }
+
+    /**
+     * 延时消息
+     *
+     * @param msgs
+     * @param consumeConcurrentlyContext
+     * @param mqExecTypeEnum
+     * @return
+     */
+    @Override
+    public ConsumeConcurrentlyStatus consumeMessageDelay(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext, MqExecTypeEnum mqExecTypeEnum) {
+        MqDto mqDto = null;
+        try {
+            long threadId = Thread.currentThread().getId();
+            String threadName = Thread.currentThread().getName();
+            for (MessageExt messageExt : msgs) {
+                log.info(":{}-:{} consumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+                log.info(":{}-:{} consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
+                int reconsumeTime = messageExt.getReconsumeTimes();
+                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
+                    this.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)) {
+                        Method method = this.getClass().getDeclaredMethod(mqExecTypeEnum.getDesc(), MqDto.class, String.class);
+                        log.info(":{}-:{} 准备执行mq exec:{}逻辑", threadId, threadName, method.getName());
+                        method.invoke(this, mqDto, SystemConstant.MQ_DELAY_TOPIC_BUFFER_LIST);
+                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                    } else {
+                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                    }
+                }
+            }
+        } catch (Exception e) {
+            log.error("mq延时消息监听，消息消费出错", e);
+            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+        } finally {
+            if (Objects.nonNull(mqDto)) {
+                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
+            }
+        }
+        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+    }
 }

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

@@ -1,22 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
 import com.qmth.themis.business.constant.SpringContextHolder;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
-import java.util.Objects;
 
 /**
  * 计算客观分
@@ -27,48 +20,10 @@ import java.util.Objects;
  */
 @Service
 public class CalculateObjectiveScoreConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(CalculateObjectiveScoreConcurrentlyImpl.class);
 
     @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
-                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
         MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.debug(":{}-:{} CalculateObjectiveScore 重试次数:{}", threadId, threadName,
-                        messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.debug(":{}-:{} CalculateObjectiveScore 接收到的消息:{}", threadId, threadName,
-                        JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
-                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
-                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
-                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.debug(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqCalculateObjectiveScoreLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 计算客观分，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_CALCULATE_OBJECTIVE_SCORE_LOGIC);
     }
 }

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

@@ -1,22 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
 import com.qmth.themis.business.constant.SpringContextHolder;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
-import java.util.Objects;
 
 /**
  * 重新算分
@@ -27,48 +20,10 @@ import java.util.Objects;
  */
 @Service
 public class CalculateScoreConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(CalculateScoreConcurrentlyImpl.class);
 
     @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
-                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
         MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.debug(":{}-:{} CalculateObjectiveScore 重试次数:{}", threadId, threadName,
-                        messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.debug(":{}-:{} CalculateObjectiveScore 接收到的消息:{}", threadId, threadName,
-                        JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
-                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
-                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
-                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.debug(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqCalculateObjectiveScoreLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 重新算分，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_CALCULATE_SCORE_LOGIC);
     }
 }

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

@@ -1,76 +1,29 @@
 package com.qmth.themis.mq.templete.impl;
 
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.Resource;
 import java.util.List;
-import java.util.Objects;
 
 /**
-* @Description: 考试断点
-* @Param:
-* @return:
-* @Author: wangliang
-* @Date: 2020/9/17
-*/
+ * @Description: 考试断点
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/9/17
+ */
 @Service
 public class ExamBreakConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(ExamBreakConcurrentlyImpl.class);
-
-    @Resource
-    RedisUtil redisUtil;
-    @Resource
-    MqLogicService mqLogicService;
 
     @Override
     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.debug(":{}-:{} exam break 重试次数:{}", threadId, threadName,
-                        messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.debug(":{}-:{} exam break 接收到的消息:{}", threadId, threadName,
-                        JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
-                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
-                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
-                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.debug(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqExamBreakLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 考试断点数据，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
+        MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_EXAM_BREAK_LOGIC);
     }
 }

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

@@ -1,22 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.Resource;
 import java.util.List;
-import java.util.Objects;
 
 /**
  * 考试记录数据初始化
@@ -27,51 +20,10 @@ import java.util.Objects;
  */
 @Service
 public class ExamRecordInitConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(ExamRecordInitConcurrentlyImpl.class);
-
-    @Resource
-    RedisUtil redisUtil;
-    @Resource
-    MqLogicService mqLogicService;
 
     @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
-                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.debug(":{}-:{} CalculateObjectiveScore 重试次数:{}", threadId, threadName,
-                        messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.debug(":{}-:{} CalculateObjectiveScore 接收到的消息:{}", threadId, threadName,
-                        JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
-                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
-                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
-                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.debug(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqRecordInitLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 考试记录数据更新，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_RECORD_INIT_LOGIC);
     }
 }

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

@@ -1,24 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
-import java.util.List;
-import java.util.Objects;
-
-import javax.annotation.Resource;
-
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
+import com.qmth.themis.mq.service.MqLogicService;
+import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
-import com.qmth.themis.mq.service.MqLogicService;
-import com.qmth.themis.mq.templete.Concurrently;
+import java.util.List;
 
 /**
  * 考试记录数据持久化
@@ -29,51 +20,10 @@ import com.qmth.themis.mq.templete.Concurrently;
  */
 @Service
 public class ExamRecordPersistedConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(ExamRecordPersistedConcurrentlyImpl.class);
-
-    @Resource
-    RedisUtil redisUtil;
-    @Resource
-    MqLogicService mqLogicService;
 
     @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
-                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.debug(":{}-:{} CalculateObjectiveScore 重试次数:{}", threadId, threadName,
-                        messageExt.getReconsumeTimes());
-                String body = new String(messageExt.getBody(), Constants.CHARSET_NAME);
-                log.debug("CalculateObjectiveScore 接收到的消息 body:{}", body);
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
-                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
-                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
-                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.debug(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqExamRecordPersistedLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 考试记录数据持久化，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_RECORD_PERSISTED_LOGIC);
     }
 }

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

@@ -1,22 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.Resource;
 import java.util.List;
-import java.util.Objects;
 
 /**
  * 考试记录数据更新
@@ -27,51 +20,10 @@ import java.util.Objects;
  */
 @Service
 public class ExamRecordUpdateConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(ExamRecordUpdateConcurrentlyImpl.class);
-
-    @Resource
-    RedisUtil redisUtil;
-    @Resource
-    MqLogicService mqLogicService;
 
     @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
-                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.debug(":{}-:{} CalculateObjectiveScore 重试次数:{}", threadId, threadName,
-                        messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.debug(":{}-:{} CalculateObjectiveScore 接收到的消息:{}", threadId, threadName,
-                        JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
-                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
-                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
-                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.debug(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqRecordUpdateLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 考试记录数据更新，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_RECORD_UPDATE_LOGIC);
     }
 }

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

@@ -1,22 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.Resource;
 import java.util.List;
-import java.util.Objects;
 
 /**
  * 考生数据更新
@@ -27,51 +20,10 @@ import java.util.Objects;
  */
 @Service
 public class ExamStudentUpdateConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(ExamStudentUpdateConcurrentlyImpl.class);
-
-    @Resource
-    RedisUtil redisUtil;
-    @Resource
-    MqLogicService mqLogicService;
 
     @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
-                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.debug(":{}-:{} CalculateObjectiveScore 重试次数:{}", threadId, threadName,
-                        messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.debug(":{}-:{} CalculateObjectiveScore 接收到的消息:{}", threadId, threadName,
-                        JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
-                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
-                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
-                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.debug(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqExamStudentUpdateLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 考生数据更新，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_EXAM_STUDENT_UPDATE_LOGIC);
     }
 }

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

@@ -1,22 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.Resource;
 import java.util.List;
-import java.util.Objects;
 
 /**
  * 人脸验证保存
@@ -27,51 +20,10 @@ import java.util.Objects;
  */
 @Service
 public class FaceVerifyConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(FaceVerifyConcurrentlyImpl.class);
-
-    @Resource
-    RedisUtil redisUtil;
-    @Resource
-    MqLogicService mqLogicService;
 
     @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
-                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.debug(":{}-:{} CalculateObjectiveScore 重试次数:{}", threadId, threadName,
-                        messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.debug(":{}-:{} CalculateObjectiveScore 接收到的消息:{}", threadId, threadName,
-                        JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
-                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
-                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
-                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.debug(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqFaceVerifySaveLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 人脸验证保存，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_FACE_VERIFY_LOGIC);
     }
 }

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

@@ -1,22 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.Resource;
 import java.util.List;
-import java.util.Objects;
 
 /**
  * 活体验证保存
@@ -27,51 +20,10 @@ import java.util.Objects;
  */
 @Service
 public class LivenessVerifyConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(LivenessVerifyConcurrentlyImpl.class);
-
-    @Resource
-    RedisUtil redisUtil;
-    @Resource
-    MqLogicService mqLogicService;
 
     @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
-                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.debug(":{}-:{} CalculateObjectiveScore 重试次数:{}", threadId, threadName,
-                        messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.debug(":{}-:{} CalculateObjectiveScore 接收到的消息:{}", threadId, threadName,
-                        JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
-                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
-                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
-                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.debug(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqLivenessVerifySaveLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 人脸验证保存，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_LIVENESS_VERIFY_LOGIC);
     }
 }

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

@@ -1,22 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
 import com.qmth.themis.business.constant.SpringContextHolder;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
-import java.util.Objects;
 
 /**
  * @Description: mq 日志并行消费监听
@@ -27,42 +20,10 @@ import java.util.Objects;
  */
 @Service
 public class LogConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(LogConcurrentlyImpl.class);
 
     @Override
     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
         MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.info(":{}-:{} Log Consumer 重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.info(":{}-:{} Log Consumer 接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.info(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqLogLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 日志，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_LOG_LOGIC);
     }
 }

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

@@ -1,15 +1,7 @@
 package com.qmth.themis.mq.templete.impl;
 
-import com.google.gson.Gson;
 import com.qmth.themis.business.constant.SpringContextHolder;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.service.TBSessionService;
-import com.qmth.themis.business.service.TMRocketMessageService;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
-import com.qmth.themis.common.exception.BusinessException;
-import com.qmth.themis.business.dto.MqDto;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
@@ -20,7 +12,6 @@ import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
-import java.util.Objects;
 
 /**
  * @Description: mq 会话并行消费监听
@@ -31,42 +22,10 @@ import java.util.Objects;
  */
 @Service
 public class SessionConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(SessionConcurrentlyImpl.class);
 
     @Override
     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
         MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.info(":{}-:{} session Consumer 重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.info(":{}-:{} session Consumer 接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.info(":{}-:{} 更新db", threadId, threadName);
-                        mqLogicService.execMqSessionLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 会话，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_SESSION_LOGIC);
     }
 }

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

@@ -1,22 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
 import com.qmth.themis.business.constant.SpringContextHolder;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
-import com.qmth.themis.business.dto.MqDto;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
-import java.util.Objects;
 
 /**
  * @Description: mq 任务并行消费监听
@@ -27,41 +20,10 @@ import java.util.Objects;
  */
 @Service
 public class TaskConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(TaskConcurrentlyImpl.class);
 
     @Override
     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
         MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.info(":{}-:{} task Consumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.info(":{}-:{} task Consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        mqLogicService.execMqTaskLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 任务，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_TASK_LOGIC);
     }
 }

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

@@ -1,22 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
 import com.qmth.themis.business.constant.SpringContextHolder;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
-import com.qmth.themis.business.dto.MqDto;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
 import com.qmth.themis.mq.service.MqLogicService;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
-import java.util.Objects;
 
 /**
  * @Description: mq 用户轨迹并行消费监听
@@ -27,42 +20,10 @@ import java.util.Objects;
  */
 @Service
 public class UserLogConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(UserLogConcurrentlyImpl.class);
 
     @Override
     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
         MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.info(":{}-:{} log Consumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
-                log.info(":{}-:{} log Consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    mqLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-                        log.info(":{}-:{} 插入用户轨迹日志", threadId, threadName);
-                        mqLogicService.execMqUserLogLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 用户轨迹，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+        return mqLogicService.consumeMessage(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_USER_LOG_LOGIC);
     }
 }

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

@@ -1,23 +1,15 @@
 package com.qmth.themis.mq.templete.impl;
 
-import java.util.List;
-import java.util.Objects;
-
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.enums.MqExecTypeEnum;
+import com.qmth.themis.mq.service.MqLogicService;
+import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
-import com.qmth.themis.business.constant.SpringContextHolder;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.dto.MqDto;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
-import com.qmth.themis.mq.service.MqLogicService;
-import com.qmth.themis.mq.templete.Concurrently;
+import java.util.List;
 
 /**
  * @Description: mq 延时消息监听 websocket超时退出 并行消费监听
@@ -28,43 +20,10 @@ import com.qmth.themis.mq.templete.Concurrently;
  */
 @Service
 public class WebsocketUnNormalConcurrentlyImpl implements Concurrently {
-    private final static Logger log = LoggerFactory.getLogger(WebsocketUnNormalConcurrentlyImpl.class);
 
     @Override
     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
         MqLogicService mqLogicService = SpringContextHolder.getBean(MqLogicService.class);
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            for (MessageExt messageExt : msgs) {
-                log.info(":{}-:{} websocket unnormal Consumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
-                String body = new String(messageExt.getBody(), Constants.CHARSET_NAME);
-                log.info("body:{}", body);
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                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)) {
-                        log.info(":{}-:{} 插入学生轨迹日志", threadId, threadName);
-                        mqLogicService.execMqWebsocketUnNormalLogic(mqDto, SystemConstant.MQ_DELAY_TOPIC_BUFFER_LIST);
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    } else {
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("mq 延时消息监听，消息消费出错", e);
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+        return mqLogicService.consumeMessageDelay(msgs, consumeConcurrentlyContext, MqExecTypeEnum.EXEC_MQ_WEBSOCKET_UN_NORMAL_LOGIC);
     }
 }

themis-task/src/main/java/com/qmth/themis/task/start/StartRunning.java

@@ -50,7 +50,7 @@ public class StartRunning implements CommandLineRunner {
         Map mqMap = new HashMap();
         mqMap.put("name", MqJob.class.getName());
         quartzService.deleteJob(QuartzTaskEnum.MQ_JOB_NAME.name(), QuartzTaskEnum.MQ_JOB_GROUP_NAME.name());
-        quartzService.addJob(MqJob.class, QuartzTaskEnum.MQ_JOB_NAME.name(), QuartzTaskEnum.MQ_JOB_GROUP_NAME.name(), "0 0/2 * * * ?", mqMap);
+        quartzService.addJob(MqJob.class, QuartzTaskEnum.MQ_JOB_NAME.name(), QuartzTaskEnum.MQ_JOB_GROUP_NAME.name(), "0 0/1 * * * ?", mqMap);
         log.info("增加mqjob end");
 
         log.info("增加mqActivityjob start");