mq修改

themis-mq/src/main/java/com/qmth/themis/mq/enums/MqTagEnum.java

@@ -24,6 +24,7 @@ public enum MqTagEnum {
     oe,
     unNormal,
     examActivity,
+    quartz,
     /**
      * 计算客观分
      */

themis-task/src/main/java/com/qmth/themis/task/listener/QuartzOrderlyImpl.java

@@ -0,0 +1,67 @@
+package com.qmth.themis.task.listener;
+
+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.mq.dto.MqDto;
+import com.qmth.themis.mq.templete.Orderly;
+import com.qmth.themis.task.listener.service.MqTaskLogicService;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
+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 quartz顺序消费监听
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/31
+ */
+@Service
+public class QuartzOrderlyImpl implements Orderly {
+    private final static Logger log = LoggerFactory.getLogger(QuartzOrderlyImpl.class);
+
+    @Override
+    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext consumeConcurrentlyContext) {
+        RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
+        MqTaskLogicService mqLogicService = SpringContextHolder.getBean(MqTaskLogicService.class);
+        MqDto mqDto = null;
+        try {
+            long threadId = Thread.currentThread().getId();
+            String threadName = Thread.currentThread().getName();
+            for (MessageExt messageExt : msgs) {
+                log.info(":{}-:{} quartz Consumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+                log.info(":{}-:{} quartz 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.execMqQuartzLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
+                        return ConsumeOrderlyStatus.SUCCESS;
+                    } else {
+                        log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
+                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+                    }
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+        } finally {
+            if (Objects.nonNull(mqDto)) {
+                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
+            }
+        }
+        return ConsumeOrderlyStatus.SUCCESS;//成功
+    }
+}

themis-task/src/main/java/com/qmth/themis/task/listener/RocketQuartzConsumer.java

@@ -1,176 +0,0 @@
-package com.qmth.themis.task.listener;
-
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONArray;
-import com.alibaba.fastjson.JSONObject;
-import com.google.gson.Gson;
-import com.qmth.themis.business.constant.SpringContextHolder;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.entity.TEExam;
-import com.qmth.themis.business.entity.TEExamActivity;
-import com.qmth.themis.business.entity.TMRocketMessage;
-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.mq.dto.MqDto;
-import com.qmth.themis.task.config.DictionaryConfig;
-import com.qmth.themis.task.quartz.ExamActivityJob;
-import com.qmth.themis.task.service.QuartzService;
-import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.client.consumer.listener.*;
-import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
-import org.apache.rocketmq.spring.annotation.SelectorType;
-import org.apache.rocketmq.spring.core.RocketMQListener;
-import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
-
-import javax.annotation.Resource;
-import java.util.*;
-
-/**
- * @Description: 普通消息监听 quartz
- * @Param:
- * @return:
- * @Author: wangliang
- * @Date: 2020/7/2
- */
-@Service
-public class RocketQuartzConsumer implements MessageListenerOrderly {
-    private final static Logger log = LoggerFactory.getLogger(RocketQuartzConsumer.class);
-
-    @Resource
-    RedisUtil redisUtil;
-
-    @Resource
-    TMRocketMessageService tmRocketMessageService;
-
-    @Resource
-    QuartzService quartzService;
-
-    @Resource
-    DictionaryConfig dictionaryConfig;
-
-    /**
-     * 考试场次监听
-     */
-    @Override
-    @Transactional
-    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext consumeOrderlyContext) {
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            Gson gson = new Gson();
-            for (MessageExt messageExt : msgs) {
-                log.info(":{}-:{} quartz Consumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.info(":{}-:{} quartz Consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
-                int reconsumeTime = messageExt.getReconsumeTimes();
-                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-                    //超过最大重试次数，保存到数据库，后续可以发短信通知系统管理人员
-                    mqDto.setAck(SystemConstant.POSION_ACK_TYPE);
-                    TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
-                    tmRocketMessageService.saveOrUpdate(tmRocketMessage);
-                    redisUtil.delete(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId());
-                } 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)) {
-                        Map<String, Object> tranMap = mqDto.getProperties();
-                        String oper = String.valueOf(tranMap.get("oper"));
-                        Object o = JacksonUtil.parseJson(tranMap.get("exam"));
-                        TEExam teExam = JSONObject.toJavaObject(JSONObject.parseObject(String.valueOf(o)), TEExam.class);
-                        mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
-                        JSONArray jsonArray = (JSONArray) JSONArray.parse(String.valueOf(mqDto.getBody()));
-                        for (int i = 0; i < jsonArray.size(); i++) {
-                            TEExamActivity teExamActivity = JSONObject.toJavaObject((JSON) jsonArray.get(i), TEExamActivity.class);
-                            if (Objects.equals("delete", oper)) {
-                                quartzService.deleteJob(teExamActivity.getCode(), dictionaryConfig.quartzConfigDomain().getExamActivityJobGroupName());
-                            } else {
-                                Integer forceFinish = teExam.getForceFinish();
-                                Date startTime = teExamActivity.getStartTime();
-                                Date finishTime = teExamActivity.getFinishTime();
-                                if (Objects.nonNull(forceFinish) && forceFinish.intValue() == 1) {//强制收卷
-                                    Integer activityMaxDurationSeconds = Objects.nonNull(teExamActivity.getMaxDurationSeconds()) ? teExamActivity.getMaxDurationSeconds() : null;
-                                    Integer maxDurationSeconds = Objects.nonNull(teExam.getMaxDurationSeconds()) ? teExam.getMaxDurationSeconds() : null;
-                                    Integer finalMaxDurationSeconds = Objects.nonNull(activityMaxDurationSeconds) ? activityMaxDurationSeconds : maxDurationSeconds;
-                                    Calendar calendar = Calendar.getInstance();
-                                    if (Objects.nonNull(finalMaxDurationSeconds)) {
-                                        calendar.setTime(startTime);
-                                        calendar.add(Calendar.SECOND, activityMaxDurationSeconds.intValue());
-                                    } else {
-                                        calendar.setTime(finishTime);
-                                    }
-                                    if (calendar.getTime().getTime() > System.currentTimeMillis()) {
-//                                    calendar.setTime(new Date());
-                                        int year = calendar.get(Calendar.YEAR);//获取年份
-                                        int month = calendar.get(Calendar.MONTH) + 1;//获取月份
-                                        int day = calendar.get(Calendar.DATE);//获取日
-                                        int hour = calendar.get(Calendar.HOUR_OF_DAY);//小时
-                                        int minute = calendar.get(Calendar.MINUTE);//分
-                                        int second = calendar.get(Calendar.SECOND);//秒
-//                                    String cron = (second + 20) + " " + (minute) + " " + hour + " " + day + " " + month + " ? " + year;
-                                        String cron = second + " " + (minute + 1) + " " + hour + " " + day + " " + month + " ? " + year;
-                                        log.info("cron:{}", cron);
-                                        String activityCode = teExamActivity.getCode();
-                                        //执行一次性延时任务
-                                        Map mapJob = new HashMap();
-                                        mapJob.put("name", activityCode);
-                                        quartzService.deleteJob(activityCode, dictionaryConfig.quartzConfigDomain().getExamActivityJobGroupName());
-                                        quartzService.addJob(SpringContextHolder.getBean(ExamActivityJob.class).getClass(), activityCode, dictionaryConfig.quartzConfigDomain().getExamActivityJobGroupName(), cron, mapJob);
-                                    }
-                                }
-                            }
-                        }
-                        TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
-                        tmRocketMessageService.saveOrUpdate(tmRocketMessage);
-                        redisUtil.delete(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId());
-                        return ConsumeOrderlyStatus.SUCCESS;
-                    } else {
-                        log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
-                    }
-                }
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-            if (e instanceof BusinessException) {
-                throw new BusinessException(e.getMessage());
-            } else {
-                throw new RuntimeException(e);
-            }
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-            }
-        }
-        return ConsumeOrderlyStatus.SUCCESS;//成功
-    }
-
-    /**
-     * quartz
-     */
-    @Service
-    @RocketMQMessageListener(consumerGroup = "${mq.config.quartzConsumerExamActivityGroup}", topic = "${mq.config.quartzTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.quartzTopicExamActivityTag}")
-    public class taskConsumerQuartzStudent implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
-
-        @Override
-        public void onMessage(Message message) {
-            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
-        }
-
-        @Override
-        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
-            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
-            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
-            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
-            defaultMQPushConsumer.registerMessageListener(RocketQuartzConsumer.this::consumeMessage);
-        }
-    }
-}

themis-task/src/main/java/com/qmth/themis/task/listener/service/MqTaskLogicService.java

@@ -0,0 +1,31 @@
+package com.qmth.themis.task.listener.service;
+
+import com.qmth.themis.mq.dto.MqDto;
+
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * @Description: mq执行逻辑
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/31
+ */
+public interface MqTaskLogicService {
+
+    /**
+     * mq最大重试次数逻辑
+     *
+     * @param mqDto
+     * @param key
+     */
+    public void execMqMaxReconsumeTime(MqDto mqDto, String key);
+
+    /**
+     * quartz逻辑
+     *
+     * @param mqDto
+     * @param key
+     */
+    public void execMqQuartzLogic(MqDto mqDto, String key) throws ClassNotFoundException, IllegalAccessException, InstantiationException, NoSuchMethodException, InvocationTargetException;
+}

themis-task/src/main/java/com/qmth/themis/task/listener/service/impl/MqTaskLogicServiceImpl.java

@@ -0,0 +1,122 @@
+package com.qmth.themis.task.listener.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.google.gson.Gson;
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.entity.TEExam;
+import com.qmth.themis.business.entity.TEExamActivity;
+import com.qmth.themis.business.entity.TMRocketMessage;
+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.mq.dto.MqDto;
+import com.qmth.themis.task.config.DictionaryConfig;
+import com.qmth.themis.task.listener.service.MqTaskLogicService;
+import com.qmth.themis.task.quartz.ExamActivityJob;
+import com.qmth.themis.task.service.QuartzService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import javax.annotation.Resource;
+import java.lang.reflect.InvocationTargetException;
+import java.util.*;
+
+/**
+ * @Description: mq执行逻辑 impl
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/31
+ */
+@Service
+public class MqTaskLogicServiceImpl implements MqTaskLogicService {
+    private final static Logger log = LoggerFactory.getLogger(MqTaskLogicServiceImpl.class);
+
+    @Resource
+    RedisUtil redisUtil;
+
+    @Resource
+    TMRocketMessageService tmRocketMessageService;
+
+    @Resource
+    QuartzService quartzService;
+
+    @Resource
+    DictionaryConfig dictionaryConfig;
+
+    /**
+     * mq最大重试次数逻辑
+     *
+     * @param mqDto
+     * @param key
+     */
+    @Override
+    @Transactional
+    public void execMqMaxReconsumeTime(MqDto mqDto, String key) {
+        //超过最大重试次数，保存到数据库，后续可以发短信通知系统管理人员
+        Gson gson = new Gson();
+        mqDto.setAck(SystemConstant.POSION_ACK_TYPE);
+        TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+        tmRocketMessageService.saveOrUpdate(tmRocketMessage);
+        redisUtil.delete(key, mqDto.getId());
+    }
+
+    @Override
+    @Transactional
+    public void execMqQuartzLogic(MqDto mqDto, String key) throws ClassNotFoundException, IllegalAccessException, InstantiationException, NoSuchMethodException, InvocationTargetException {
+        Gson gson = new Gson();
+        Map<String, Object> tranMap = mqDto.getProperties();
+        String oper = String.valueOf(tranMap.get("oper"));
+        Object o = JacksonUtil.parseJson(tranMap.get("exam"));
+        TEExam teExam = JSONObject.toJavaObject(JSONObject.parseObject(String.valueOf(o)), TEExam.class);
+        mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
+        JSONArray jsonArray = (JSONArray) JSONArray.parse(String.valueOf(mqDto.getBody()));
+        for (int i = 0; i < jsonArray.size(); i++) {
+            TEExamActivity teExamActivity = JSONObject.toJavaObject((JSON) jsonArray.get(i), TEExamActivity.class);
+            if (Objects.equals("delete", oper)) {
+                quartzService.deleteJob(teExamActivity.getCode(), dictionaryConfig.quartzConfigDomain().getExamActivityJobGroupName());
+            } else {
+                Integer forceFinish = teExam.getForceFinish();
+                Date startTime = teExamActivity.getStartTime();
+                Date finishTime = teExamActivity.getFinishTime();
+                if (Objects.nonNull(forceFinish) && forceFinish.intValue() == 1) {//强制收卷
+                    Integer activityMaxDurationSeconds = Objects.nonNull(teExamActivity.getMaxDurationSeconds()) ? teExamActivity.getMaxDurationSeconds() : null;
+                    Integer maxDurationSeconds = Objects.nonNull(teExam.getMaxDurationSeconds()) ? teExam.getMaxDurationSeconds() : null;
+                    Integer finalMaxDurationSeconds = Objects.nonNull(activityMaxDurationSeconds) ? activityMaxDurationSeconds : maxDurationSeconds;
+                    Calendar calendar = Calendar.getInstance();
+                    if (Objects.nonNull(finalMaxDurationSeconds)) {
+                        calendar.setTime(startTime);
+                        calendar.add(Calendar.SECOND, (activityMaxDurationSeconds.intValue() + 60));
+                    } else {
+                        calendar.setTime(finishTime);
+                    }
+                    if (calendar.getTime().getTime() > System.currentTimeMillis()) {
+//                        calendar.setTime(new Date());
+                        int year = calendar.get(Calendar.YEAR);//获取年份
+                        int month = calendar.get(Calendar.MONTH) + 1;//获取月份
+                        int day = calendar.get(Calendar.DATE);//获取日
+                        int hour = calendar.get(Calendar.HOUR_OF_DAY);//小时
+                        int minute = calendar.get(Calendar.MINUTE);//分
+                        int second = calendar.get(Calendar.SECOND);//秒
+                        String cron = second + " " + minute + " " + hour + " " + day + " " + month + " ? " + year;
+                        log.info("cron:{}", cron);
+                        String activityCode = teExamActivity.getCode();
+                        //执行一次性延时任务
+                        Map mapJob = new HashMap();
+                        mapJob.put("name", activityCode);
+                        quartzService.deleteJob(activityCode, dictionaryConfig.quartzConfigDomain().getExamActivityJobGroupName());
+                        quartzService.addJob(SpringContextHolder.getBean(ExamActivityJob.class).getClass(), activityCode, dictionaryConfig.quartzConfigDomain().getExamActivityJobGroupName(), cron, mapJob);
+                    }
+                }
+            }
+        }
+        TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+        tmRocketMessageService.saveOrUpdate(tmRocketMessage);
+        redisUtil.delete(key, mqDto.getId());
+    }
+}

themis-task/src/main/java/com/qmth/themis/task/quartz/ExamActivityJob.java

@@ -15,8 +15,8 @@ import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.BeanUtils;
 import org.springframework.scheduling.quartz.QuartzJobBean;
+import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import javax.annotation.Resource;
@@ -32,6 +32,7 @@ import java.util.Objects;
  * @Author: wangliang
  * @Date: 2020/7/27
  */
+@Service
 public class ExamActivityJob extends QuartzJobBean {
     private final static Logger log = LoggerFactory.getLogger(ExamActivityJob.class);
 
@@ -76,7 +77,7 @@ public class ExamActivityJob extends QuartzJobBean {
                     });
                     tOeExamRecordService.updateBatchById(tOeExamRecordList);
 
-                    if (examStudentIdList.size() > 0) {
+                    if (Objects.nonNull(examStudentIdList) && examStudentIdList.size() > 0) {
                         //获取该考试批次下所有考生，考试次数减1
                         QueryWrapper<TEExamStudent> teExamStudentQueryWrapper = new QueryWrapper<>();
                         teExamStudentQueryWrapper.lambda().in(TEExamStudent::getId, examStudentIdList);

themis-task/src/main/java/com/qmth/themis/task/quartz/MqJob.java

@@ -1,7 +1,6 @@
 package com.qmth.themis.task.quartz;
 
 import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.threadPool.MyThreadPool;
 import com.qmth.themis.business.util.RedisUtil;
 import com.qmth.themis.mq.dto.MqDto;
 import com.qmth.themis.mq.service.ProducerServer;
@@ -10,6 +9,7 @@ import org.quartz.JobExecutionException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.scheduling.quartz.QuartzJobBean;
+import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
 import java.util.Map;
@@ -22,6 +22,7 @@ import java.util.Objects;
  * @Author: wangliang
  * @Date: 2020/7/1
  */
+@Service
 public class MqJob extends QuartzJobBean {
     private final static Logger log = LoggerFactory.getLogger(MqJob.class);
 
@@ -31,19 +32,12 @@ public class MqJob extends QuartzJobBean {
     @Resource
     ProducerServer producerServer;
 
-    @Resource
-    MyThreadPool myThreadPool;
-
     @Override
     protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
-        myThreadPool.arbitratePoolTaskExecutor.execute(() -> {
-            log.info("mq_job进来了,context:{}", context);
-            this.assembleJob(SystemConstant.MQ_TOPIC_BUFFER_LIST);
-        });
-        myThreadPool.arbitratePoolTaskExecutor.execute(() -> {
-            log.info("mq_delay_job进来了,context:{}", context);
-            this.assembleDelayJob(SystemConstant.MQ_DELAY_TOPIC_BUFFER_LIST);
-        });
+        log.info("mq_job进来了,context:{}", context);
+        this.assembleJob(SystemConstant.MQ_TOPIC_BUFFER_LIST);
+        log.info("mq_delay_job进来了,context:{}", context);
+        this.assembleDelayJob(SystemConstant.MQ_DELAY_TOPIC_BUFFER_LIST);
     }
 
     /**

themis-task/src/main/java/com/qmth/themis/task/service/impl/QuartzServiceImpl.java

@@ -16,7 +16,7 @@ import java.util.*;
  * @Author: wangliang
  * @Date: 2020/7/1
  */
-@Service
+@Service("quartzService")
 public class QuartzServiceImpl implements QuartzService {
 
     @Resource

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

@@ -7,6 +7,7 @@ import com.qmth.themis.mq.enums.MqTopicEnum;
 import com.qmth.themis.mq.listener.RocketMessageConsumer;
 import com.qmth.themis.mq.templete.impl.*;
 import com.qmth.themis.task.config.DictionaryConfig;
+import com.qmth.themis.task.listener.QuartzOrderlyImpl;
 import com.qmth.themis.task.quartz.MqJob;
 import com.qmth.themis.task.service.QuartzService;
 import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
@@ -95,6 +96,14 @@ public class StartRunning implements CommandLineRunner {
          */
         //计算客观分
         rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.normalGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.calculateObjectiveScore.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(CalculateObjectiveScoreConcurrentlyImpl.class));
+
+        /**
+         * quartz mq start
+         */
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.quartzConsumerExamActivityGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.examActivity.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(QuartzOrderlyImpl.class));
+        /**
+         * quartz mq end
+         */
         log.info("服务器启动时执行 end");
     }
 }

themis-task/src/main/resources/application.properties

@@ -172,7 +172,7 @@ rocketmq.producer.enable-msg-trace=true
 #\u81EA\u5B9A\u4E49\u7684\u6D88\u606F\u8F68\u8FF9\u4E3B\u9898
 rocketmq.producer.customized-trace-topic=my-trace-topic
 
-#quartz-\u8003\u8BD5\u573A\u6B21task
-mq.config.quartzTopic=themis-topic-exam
-mq.config.quartzConsumerExamActivityGroup=themis-group-exam-quartzExamActivity
-mq.config.quartzTopicExamActivityTag=examActivity
+##quartz-\u8003\u8BD5\u573A\u6B21task
+#mq.config.quartzTopic=themis-topic-exam
+#mq.config.quartzConsumerExamActivityGroup=themis-group-exam-quartzExamActivity
+#mq.config.quartzTopicExamActivityTag=examActivity