考务导入接口

themis-backend/src/main/java/com/qmth/themis/backend/api/SysController.java

@@ -9,7 +9,6 @@ import com.qmth.themis.business.entity.TBAttachment;
 import com.qmth.themis.business.entity.TBPrivilege;
 import com.qmth.themis.business.entity.TBUser;
 import com.qmth.themis.business.enums.RoleEnum;
-import com.qmth.themis.business.enums.UploadFileEnum;
 import com.qmth.themis.business.service.TBAttachmentService;
 import com.qmth.themis.business.service.TBPrivilegeService;
 import com.qmth.themis.common.enums.ExceptionResultEnum;
@@ -91,7 +90,7 @@ public class SysController {
     @RequestMapping(value = "/file/upload", method = RequestMethod.POST)
     @Transactional
     @ApiResponses({@ApiResponse(code = 200, message = "{\"id\":0}", response = Result.class)})
-    public Result fileUpload(@ApiParam(value = "上传文件", required = true) @RequestParam MultipartFile file, @RequestParam Integer type) throws IOException {
+    public Result fileUpload(@ApiParam(value = "上传文件", required = true) @RequestParam MultipartFile file, @ApiParam(value = "上传文件类型", required = true) @RequestParam Integer type) throws IOException {
         if (Objects.isNull(file) || Objects.equals(file, "")) {
             throw new BusinessException(ExceptionResultEnum.ATTACHMENT_IS_NULL);
         }

themis-backend/src/main/java/com/qmth/themis/backend/api/TBExamInvigilateUserController.java

@@ -72,11 +72,10 @@ public class TBExamInvigilateUserController {
             TBOrg tbOrg = (TBOrg) ServletUtil.getRequestOrg();
             QueryWrapper<TBExamInvigilateUser> tbExamInvigilateUserQueryWrapper = new QueryWrapper<>();
             tbExamInvigilateUserQueryWrapper.lambda().eq(TBExamInvigilateUser::getRoomCode, roomCode);
-            Long orgId = null;
-            if (Objects.nonNull(tbOrg)) {
-                tbExamInvigilateUserQueryWrapper.lambda().eq(TBExamInvigilateUser::getOrgId, tbOrg.getId());
-                orgId = tbOrg.getId();
+            if (Objects.nonNull(tbOrg) && Objects.isNull(tbOrg.getId())) {
+                throw new BusinessException("监考员机构id不允许为空");
             }
+            Long orgId = tbOrg.getId();
             List<TBExamInvigilateUser> tbExamInvigilateUserList = tbExamInvigilateUserService.list(tbExamInvigilateUserQueryWrapper);
             List<String> roomNameList = tbExamInvigilateUserList.stream().map(s -> s.getRoomName()).collect(Collectors.toList());
             tbExamInvigilateUserService.remove(tbExamInvigilateUserQueryWrapper);
@@ -107,7 +106,7 @@ public class TBExamInvigilateUserController {
     @RequestMapping(value = "/import", method = RequestMethod.POST)
     @Transactional
     @ApiResponses({@ApiResponse(code = 200, message = "{\"taskId\":0}", response = Result.class)})
-    public Result importData(@ApiParam(value = "上传文件", required = true) @RequestParam MultipartFile file, @RequestParam Integer type) {
+    public Result importData(@ApiParam(value = "上传文件", required = true) @RequestParam MultipartFile file, @ApiParam(value = "上传文件类型", required = true) @RequestParam Integer type) {
         if (Objects.isNull(file) || Objects.equals(file, "")) {
             throw new BusinessException(ExceptionResultEnum.ATTACHMENT_IS_NULL);
         }

themis-backend/src/main/java/com/qmth/themis/backend/api/TEExamController.java

@@ -61,6 +61,9 @@ public class TEExamController {
             TBUser tbUser = (TBUser) ServletUtil.getRequestAccount();
             TBOrg tbOrg = (TBOrg) ServletUtil.getRequestOrg();
             if (Objects.nonNull(tbOrg)) {
+                if (Objects.isNull(tbOrg.getId())) {
+                    throw new BusinessException("考试的机构id不允许为空");
+                }
                 teExam.setOrgId(tbOrg.getId());
             }
             if (Objects.nonNull(teExam.getId())) {

themis-backend/src/main/java/com/qmth/themis/backend/api/TEExamPaperController.java

@@ -96,7 +96,7 @@ public class TEExamPaperController {
     @RequestMapping(value = "/import", method = RequestMethod.POST)
     @Transactional
     @ApiResponses({@ApiResponse(code = 200, message = "{\"taskId\":0}", response = Result.class)})
-    public Result importData(@ApiParam(value = "上传文件", required = true) @RequestParam MultipartFile file, @RequestParam Integer type) {
+    public Result importData(@ApiParam(value = "上传文件", required = true) @RequestParam MultipartFile file, @ApiParam(value = "上传文件类型", required = true) @RequestParam Integer type) {
         if (Objects.isNull(file) || Objects.equals(file, "")) {
             throw new BusinessException(ExceptionResultEnum.ATTACHMENT_IS_NULL);
         }

themis-backend/src/main/java/com/qmth/themis/backend/api/TEExamStudentController.java

@@ -1,33 +1,30 @@
 package com.qmth.themis.backend.api;
 
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.qmth.themis.backend.config.DictionaryConfig;
 import com.qmth.themis.backend.constant.BackendConstant;
 import com.qmth.themis.backend.util.ServletUtil;
 import com.qmth.themis.business.annotation.ApiJsonObject;
 import com.qmth.themis.business.annotation.ApiJsonProperty;
 import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.entity.TBAttachment;
-import com.qmth.themis.business.entity.TBTaskHistory;
-import com.qmth.themis.business.entity.TBUser;
-import com.qmth.themis.business.entity.TEExamStudent;
+import com.qmth.themis.business.entity.*;
+import com.qmth.themis.business.enums.MqEnum;
 import com.qmth.themis.business.enums.TaskTypeEnum;
 import com.qmth.themis.business.enums.TaskTypeExecEnum;
-import com.qmth.themis.business.service.TBAttachmentService;
-import com.qmth.themis.business.service.TBTaskHistoryService;
-import com.qmth.themis.business.service.TEExamStudentService;
-import com.qmth.themis.business.templete.TaskImportTemplete;
-import com.qmth.themis.business.templete.impl.ExamStudentTemplete;
+import com.qmth.themis.business.service.*;
 import com.qmth.themis.business.threadPool.MyThreadPool;
 import com.qmth.themis.business.util.JacksonUtil;
 import com.qmth.themis.common.enums.ExceptionResultEnum;
 import com.qmth.themis.common.exception.BusinessException;
 import com.qmth.themis.common.util.Result;
 import com.qmth.themis.common.util.ResultUtil;
+import com.qmth.themis.mq.dto.MqDto;
+import com.qmth.themis.mq.service.MqDtoService;
 import io.swagger.annotations.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.dao.DuplicateKeyException;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.web.bind.annotation.*;
 import org.springframework.web.multipart.MultipartFile;
@@ -35,6 +32,8 @@ import org.springframework.web.multipart.MultipartFile;
 import javax.annotation.Resource;
 import java.io.IOException;
 import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /**
  * @Description: 考生库 前端控制器
@@ -125,19 +124,38 @@ public class TEExamStudentController {
     @Resource
     MyThreadPool myThreadPool;
 
+    @Resource
+    TEExamService teExamService;
+
+    @Resource
+    TEExamActivityService teExamActivityService;
+
+    @Resource
+    TBExamInvigilateUserService tbExamInvigilateUserService;
+
+    @Resource
+    MqDtoService mqDtoService;
+
+    @Resource
+    DictionaryConfig dictionaryConfig;
+
     @ApiOperation(value = "考生导入接口")
     @RequestMapping(value = "/import", method = RequestMethod.POST)
     @Transactional
     @ApiResponses({@ApiResponse(code = 200, message = "{\"taskId\":0}", response = Result.class)})
-    public Result importData(@ApiParam(value = "上传文件", required = true) @RequestParam MultipartFile file, @RequestParam Integer type) throws IOException {
+    public Result importData(@ApiParam(value = "上传文件", required = true) @RequestParam MultipartFile file, @ApiParam(value = "上传文件类型", required = true) @RequestParam Integer type, @ApiParam(value = "考试批次id", required = true) @RequestParam Long examId) throws IOException {
         if (Objects.isNull(file) || Objects.equals(file, "")) {
             throw new BusinessException(ExceptionResultEnum.ATTACHMENT_IS_NULL);
         }
+        if (Objects.isNull(examId) || Objects.equals(examId, "")) {
+            throw new BusinessException(ExceptionResultEnum.EXAM_ID_IS_NULL);
+        }
         log.info("开始考生导入");
         long start = System.currentTimeMillis();
         Map<String, Object> mapParameter = BackendConstant.getAttachmentEnv(type);
         TBAttachment tbAttachment = null;
         TBTaskHistory tbTaskHistory = null;
+        Map transMap = new HashMap();
         try {
             TBUser tbUser = (TBUser) ServletUtil.getRequestAccount();
             tbAttachment = tbAttachmentService.saveAttachment(file, ServletUtil.getRequestMd5(), ServletUtil.getRequestPath(), mapParameter, tbUser.getOrgId(), tbUser.getId());
@@ -145,27 +163,47 @@ public class TEExamStudentController {
                 throw new BusinessException(ExceptionResultEnum.ATTACHMENT_ERROR);
             } else {
                 //往任务表里插一条数据
-                tbTaskHistory = new TBTaskHistory();
-                tbTaskHistory.setType(TaskTypeEnum.import_exam_student.ordinal());
-                tbTaskHistory.setEntityId(tbAttachment.getId());
-                tbTaskHistory.setStatus(TaskTypeExecEnum.init.ordinal());
-                tbTaskHistory.setProgress(0d);
-                tbTaskHistory.setImportFileName(tbAttachment.getName());
-                tbTaskHistory.setImportFilePath(tbAttachment.getPath());
-                tbTaskHistory.setCreateId(tbUser.getId());
-                tbTaskHistory.setStartTime(new Date());
+                tbTaskHistory = new TBTaskHistory(TaskTypeEnum.import_exam_student.ordinal(), tbAttachment.getId(), TaskTypeExecEnum.init.ordinal(), "准备开始处理", 0d, tbAttachment.getName(), tbAttachment.getRemark(), tbUser.getId());
                 taskHistoryService.save(tbTaskHistory);
+                transMap.put("tbTaskHistory", tbTaskHistory);
+            }
+            transMap.put("examId", examId);
+            transMap.put("createId", tbUser.getId());
+
+            //先查询考试相关信息
+            TEExam teExam = teExamService.getById(examId);
+            if (Objects.isNull(teExam)) {
+                throw new BusinessException(ExceptionResultEnum.EXAM_NO);
+            }
+            transMap.put("mode", teExam.getMode());
+            transMap.put("orgId", teExam.getOrgId());
+
+            //获取该批次下的所有考场，校验考场id是否存在
+            QueryWrapper<TEExamActivity> teExamActivityQueryWrapper = new QueryWrapper<>();
+            teExamActivityQueryWrapper.lambda().eq(TEExamActivity::getExamId, examId);
+            List<TEExamActivity> teExamActivityList = teExamActivityService.list(teExamActivityQueryWrapper);
+            if (Objects.isNull(teExamActivityList) || teExamActivityList.size() == 0) {
+                throw new BusinessException(ExceptionResultEnum.EXAM_ACTIVITY_NO);
+            }
+            Map<String, TEExamActivity> teExamActivityMap = teExamActivityList.stream().collect(Collectors.toMap(TEExamActivity::getCode, Function.identity(), (dto1, dto2) -> dto1));
+            transMap.put("teExamActivityMap", teExamActivityMap);
+
+            //获取考场代码和考场名称
+            QueryWrapper<TBExamInvigilateUser> tbExamInvigilateUserQueryWrapper = new QueryWrapper<>();
+            tbExamInvigilateUserQueryWrapper.lambda().eq(TBExamInvigilateUser::getOrgId, teExam.getOrgId());
+            List<TBExamInvigilateUser> tbExamInvigilateUserList = tbExamInvigilateUserService.list(tbExamInvigilateUserQueryWrapper);
+            if (Objects.nonNull(tbExamInvigilateUserList) && tbExamInvigilateUserList.size() > 0) {
+                Map<String, String> tbExamInvigilateUserMap = new HashMap();
+                tbExamInvigilateUserList.forEach(s -> {
+                    tbExamInvigilateUserMap.put(s.getRoomCode() + ":" + s.getRoomName(), s.getRoomCode() + ":" + s.getRoomName());
+                });
+                transMap.put("tbExamInvigilateUserMap", tbExamInvigilateUserMap);
             }
-            //异步插入数据
-            TBTaskHistory finalTbTaskHistory = tbTaskHistory;
-            myThreadPool.arbitratePoolTaskExecutor.submit(() -> {
-                TaskImportTemplete examStudentTemplete = new ExamStudentTemplete();
-                try {
-                    examStudentTemplete.importTask(file, finalTbTaskHistory);
-                } catch (IOException e) {
-                    e.printStackTrace();
-                }
-            });
+            transMap.put("remark", tbAttachment.getRemark());
+            //mq发送消息start
+            MqDto mqDto = new MqDto(dictionaryConfig.mqConfigDomain().getTaskTopic(), dictionaryConfig.mqConfigDomain().getTaskTopicExamStudentTag(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
+            mqDtoService.assembleSendOneWayMsg(mqDto);
+            //mq发送消息end
             long end = System.currentTimeMillis();
             log.info("考生导入结束,============耗时============:{}秒", (end - start) / 1000);
         } catch (Exception e) {
@@ -173,12 +211,7 @@ public class TEExamStudentController {
             if (Objects.nonNull(tbAttachment)) {
                 tbAttachmentService.deleteAttachment(mapParameter, tbAttachment);
             }
-            if (e instanceof DuplicateKeyException) {
-//                String errorColumn = e.getCause().toString();
-//                String column = errorColumn.substring(errorColumn.indexOf("'") + 1, errorColumn.lastIndexOf("-"));
-//                String columnStr = errorColumn.substring(errorColumn.lastIndexOf("key") + 3, errorColumn.length()).replaceAll("'", "");
-//                throw new BusinessException(AttachmentErrorEnum.convertToCode(columnStr) + "-" + column + ",数据不允许重复插入");
-            } else if (e instanceof BusinessException) {
+            if (e instanceof BusinessException) {
                 throw new BusinessException(e.getMessage());
             } else {
                 throw new RuntimeException(e);

themis-backend/src/main/java/com/qmth/themis/backend/listener/RocketSessionConsumer.java

@@ -0,0 +1,235 @@
+package com.qmth.themis.backend.listener;
+
+import com.google.gson.Gson;
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.entity.TBSession;
+import com.qmth.themis.business.entity.TMRocketMessage;
+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.mq.dto.MqDto;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+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.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @Description: 普通消息监听 session_topic
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/6/28
+ */
+@Service
+public class RocketSessionConsumer implements
+//        MessageListenerOrderly
+        MessageListenerConcurrently //并发消费
+{
+
+    private final static Logger log = LoggerFactory.getLogger(RocketSessionConsumer.class);
+
+    @Resource
+    TBSessionService tbSessionService;
+
+    @Resource
+    RedisUtil redisUtil;
+
+    @Resource
+    TMRocketMessageService tmRocketMessageService;
+
+    /**
+     * 并发消费
+     *
+     * @param msgs
+     * @param consumeConcurrentlyContext
+     * @return
+     */
+    @Override
+    @Transactional
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        MqDto mqDto = null;
+        try {
+            long threadId = Thread.currentThread().getId();
+            String threadName = Thread.currentThread().getName();
+            Gson gson = new Gson();
+            for (MessageExt messageExt : msgs) {
+                log.info(":{}-:{} sessionConsumer 重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+                log.info(":{}-:{} sessionConsumer 接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
+                log.info(":{}-:{} sessionConsumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
+                if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
+                    log.info(":{}-:{} 更新db", threadId, threadName);
+                    tbSessionService.saveSessionInfo(JacksonUtil.readJson(JacksonUtil.parseJson(mqDto.getBody()), TBSession.class), mqDto.getTimestamp());
+                    mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
+                    TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+                    tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
+                    tmRocketMessageService.saveOrUpdate(tmRocketMessage);
+                    redisUtil.deleteMqTopicList(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId());
+                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                } else {
+                    log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
+                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            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 consumeOrderlyContext
+//     * @return
+//     */
+//    @Override
+//    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext
+//            consumeOrderlyContext) {
+//        try {
+//            for (MessageExt messageExt : msgs) {
+//                log.info("sessionConsumer重试次数:{}", messageExt.getReconsumeTimes());
+//                MqDto mqDto = (MqDto) toJavaObject(parseObject(new String(messageExt.getBody(), Constants.CHARSET)), MqDto.class);
+//                log.info("sessionConsumer接受到的消息:{}", JacksonUtil.parseJson(mqDto));
+//                log.info("mqDto sequence:{},tag:{}", mqDto.getSequence(), mqDto.getTag());
+//                MqDto redisMqdto = (MqDto) redisUtil.getSessionTopicList(mqDto.getId());
+//                if (Objects.nonNull(redisMqdto)) {
+//                    if (Objects.nonNull(redisMqdto.getAck()) && redisMqdto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
+//                        log.info("更新db");
+//                        tbSessionService.saveSessionInfo(toJavaObject((JSON) mqDto.getBody(), TBSession.class), redisMqdto.getTimestamp());
+//                        redisUtil.deleteSessionTopicList(redisMqdto.getId());
+//                        return ConsumeOrderlyStatus.SUCCESS;
+//                    } else {
+//                        log.info("消息ack未确认，重发");
+//                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+//                    }
+//                } else {
+//                    log.info("消息数据为空，重发消息");
+//                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+//                }
+//            }
+//        } catch (Exception e) {
+//            e.printStackTrace();
+//            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+//        }
+//        return ConsumeOrderlyStatus.SUCCESS;//成功
+//    }
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWebGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWebTag}")
+    public class sessionConsumerWeb 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.setMessageModel(MessageModel.BROADCASTING);
+            defaultMQPushConsumer.registerMessageListener(RocketSessionConsumer.this::consumeMessage);
+        }
+    }
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerPcGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicPcTag}")
+    public class sessionConsumerPc 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(RocketSessionConsumer.this::consumeMessage);
+        }
+    }
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWxappMonitorGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWxappMonitorTag}")
+    public class sessionConsumerWxappMonitor 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(RocketSessionConsumer.this::consumeMessage);
+        }
+    }
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWxappAnswerGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWxappAnswerTag}")
+    public class sessionConsumerWxappAnswer 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(RocketSessionConsumer.this::consumeMessage);
+        }
+    }
+
+    /**
+     * 死信队列
+     */
+//    @Service
+//    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerGroupDlq}", topic = "${mq.config.sessionTopicDlq}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicTag}")
+//    public class dlqSessionConsumer implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+//
+//        @Override
+//        public void onMessage(Message message) {
+//            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
+//        }
+//
+//        @Override
+//        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
+//            log.info("dlqSessionConsumer死信队列进来了");
+//            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
+//            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
+//            defaultMQPushConsumer.registerMessageListener(RocketConsumer.this::consumeMessage);
+//        }
+//    }
+}

themis-backend/src/main/java/com/qmth/themis/backend/listener/RocketUserLogConsumer.java

@@ -0,0 +1,128 @@
+package com.qmth.themis.backend.listener;
+
+import com.google.gson.Gson;
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.entity.TMRocketMessage;
+import com.qmth.themis.business.enums.MqEnum;
+import com.qmth.themis.business.enums.SystemOperationEnum;
+import com.qmth.themis.business.service.TEUserLogService;
+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.mq.dto.MqDto;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+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.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @Description: 普通消息监听 用户日志
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/2
+ */
+@Service
+public class RocketUserLogConsumer implements MessageListenerConcurrently {
+    private final static Logger log = LoggerFactory.getLogger(RocketUserLogConsumer.class);
+
+    @Resource
+    RedisUtil redisUtil;
+
+    @Resource
+    TEUserLogService teUserLogService;
+
+    @Resource
+    TMRocketMessageService tmRocketMessageService;
+
+    @Override
+    @Transactional
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        MqDto mqDto = null;
+        try {
+            long threadId = Thread.currentThread().getId();
+            String threadName = Thread.currentThread().getName();
+            Gson gson = new Gson();
+            for (MessageExt messageExt : msgs) {
+                log.info(":{}-:{} userLogConsumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+//                MqDto mqDto = (MqDto) toJavaObject(parseObject(new String(messageExt.getBody(), Constants.CHARSET)), MqDto.class);
+                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+                log.info(":{}-:{} userLogConsumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
+                log.info(":{}-:{} userLogConsumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
+                if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
+                    log.info(":{}-:{} 插入用户轨迹日志", threadId, threadName);
+                    teUserLogService.saveUserLogInfo(mqDto.getTimestamp(), mqDto.getObjId(), MqEnum.valueOf(String.valueOf(mqDto.getType())).getId(), SystemOperationEnum.valueOf(String.valueOf(mqDto.getBody())).getCode(), JacksonUtil.parseJson(mqDto));
+                    mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
+                    TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+                    tmRocketMessageService.saveOrUpdate(tmRocketMessage);
+                    redisUtil.deleteMqTopicList(SystemConstant.USERLOG_TOPIC_BUFFER_LIST, mqDto.getId());
+                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                } else {
+                    log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
+                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+        } finally {
+            if (Objects.nonNull(mqDto)) {
+                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId());
+            }
+        }
+        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+    }
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.userLogConsumerUserGroup}", topic = "${mq.config.userLogTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.userLogTopicUserTag}")
+    public class sessionConsumerUserLog 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(RocketUserLogConsumer.this::consumeMessage);
+        }
+    }
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.userLogConsumerStudentGroup}", topic = "${mq.config.userLogTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.userLogTopicStudentTag}")
+    public class sessionConsumerStudentLog 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(RocketUserLogConsumer.this::consumeMessage);
+        }
+    }
+}

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

@@ -162,8 +162,8 @@ mq.config.sessionConsumerGroup=${mq.config.server}-group-session
 mq.config.sessionTopicWebTag=web
 mq.config.sessionConsumerWebGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicWebTag}
 
-mq.config.sessionTopicWxappVideoTag=wxapp_video
-mq.config.sessionConsumerWxappVideoGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicWxappVideoTag}
+mq.config.sessionTopicWxappMonitorTag=wxapp_monitor
+mq.config.sessionConsumerWxappMonitorGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicWxappMonitorTag}
 
 mq.config.sessionTopicWxappAnswerTag=wxapp_answer
 mq.config.sessionConsumerWxappAnswerGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicWxappAnswerTag}
@@ -181,6 +181,13 @@ mq.config.userLogConsumerUserGroup=${mq.config.userLogConsumerGroup}-${mq.config
 mq.config.userLogTopicStudentTag=student
 mq.config.userLogConsumerStudentGroup=${mq.config.userLogConsumerGroup}-${mq.config.userLogTopicStudentTag}
 
+#task_topic\u76D1\u542C
+mq.config.taskTopic=${mq.config.server}-topic-task
+mq.config.taskConsumerGroup=${mq.config.server}-group-task
+
+mq.config.taskTopicExamStudentTag=examStudent
+mq.config.taskConsumerExamStudentGroup=${mq.config.taskConsumerGroup}-${mq.config.taskTopicExamStudentTag}
+
 #dlq\u6B7B\u4FE1\u961F\u5217
 #mq.config.sessionConsumerGroupDlq=${mq.config.sessionConsumerGroup}-dlq
 #mq.config.sessionTopicDlq=%DLQ%${mq.config.sessionConsumerGroup}

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

@@ -46,6 +46,7 @@ public class SystemConstant {
     public static final String ERROR = "/error";
     public static final String SESSION_TOPIC_BUFFER_LIST = "session:topic:buffer:list";
     public static final String USERLOG_TOPIC_BUFFER_LIST = "userLog:topic:buffer:list";
+    public static final String TASKLOG_TOPIC_BUFFER_LIST = "taskLog:topic:buffer:list";
     public static final String TYPE = "type";
     public static final String LOCAL = "local";
     public static final String OSS = "oss";
@@ -55,7 +56,8 @@ public class SystemConstant {
     public static final String MD5 = "md5";
     public static final String ATTACHMENT_TYPE = "attachmentType";
     public static final String UPLOAD_TYPE = "uploadType";
-    public static final String DEFAULT_PASSWORD = "123456";
+    public static final String DEFAULT_PASSWORD = "yXVUkR45PFz0UfpbDB8/ew==";
+    public static final String USER_DIR = "user.dir";
     /**
      * session过期时间
      */
@@ -63,6 +65,11 @@ public class SystemConstant {
     public static final int PC_SESSION_EXPIRE = 1;//过期时间1天
     public static final int WXAPP_VIDEO_SESSION_EXPIRE = 30;//过期时间30天
     public static final int WXAPP_ANSWER_SESSION_EXPIRE = 30;//过期时间30天
+    /**
+     * redis分布式锁
+     */
+    public static final String REDIS_LOCK_MQ_PREFIX = "lock:mq:";
+    public static final long REDIS_LOCK_MQ_TIME_OUT = 1L;
     /**
      * redis过期时间
      */
@@ -107,7 +114,7 @@ public class SystemConstant {
             calendar.add(Calendar.DAY_OF_YEAR, SystemConstant.WEB_SESSION_EXPIRE);
         } else if (Objects.equals(platform.getSource(), Source.wxapp_answer.name())) {
             calendar.add(Calendar.DAY_OF_YEAR, SystemConstant.WXAPP_ANSWER_SESSION_EXPIRE);
-        } else if (Objects.equals(platform.getSource(), Source.wxapp_video.name())) {
+        } else if (Objects.equals(platform.getSource(), Source.wxapp_monitor.name())) {
             calendar.add(Calendar.DAY_OF_YEAR, SystemConstant.WXAPP_VIDEO_SESSION_EXPIRE);
         } else if (Objects.equals(platform.getSource(), Source.pc.name())) {
             calendar.add(Calendar.DAY_OF_YEAR, SystemConstant.PC_SESSION_EXPIRE);

themis-business/src/main/java/com/qmth/themis/business/domain/MqConfigDomain.java

@@ -14,15 +14,21 @@ public class MqConfigDomain {
     private String sessionConsumerGroup;
     //    private String sessionConsumerGroupDlq;
 //    private String sessionTopicDlq;
+    /**taskTopicExamStudentTag
+     * session topic
+     */
     private String sessionTopicWebTag;
-    private String sessionTopicWxappVideoTag;
+    private String sessionTopicWxappMonitorTag;
     private String sessionTopicWxappAnswerTag;
     private String sessionTopicPcTag;
     private String sessionConsumerWebGroup;
-    private String sessionConsumerWxappVideoGroup;
+    private String sessionConsumerWxappMonitorGroup;
     private String sessionConsumerWxappAnswerGroup;
     private String sessionConsumerPcGroup;
 
+    /**
+     * user topic
+     */
     private String userLogTopic;
     private String userLogConsumerGroup;
     private String userLogTopicUserTag;
@@ -30,12 +36,44 @@ public class MqConfigDomain {
     private String userLogTopicStudentTag;
     private String userLogConsumerStudentGroup;
 
-    public String getSessionTopicWxappVideoTag() {
-        return sessionTopicWxappVideoTag;
+    /**
+     * task group
+     */
+    private String taskTopic;
+    private String taskConsumerGroup;
+    private String taskTopicExamStudentTag;
+    private String taskConsumerExamStudentGroup;
+
+    public String getTaskTopic() {
+        return taskTopic;
+    }
+
+    public void setTaskTopic(String taskTopic) {
+        this.taskTopic = taskTopic;
+    }
+
+    public String getTaskConsumerGroup() {
+        return taskConsumerGroup;
+    }
+
+    public void setTaskConsumerGroup(String taskConsumerGroup) {
+        this.taskConsumerGroup = taskConsumerGroup;
+    }
+
+    public String getTaskTopicExamStudentTag() {
+        return taskTopicExamStudentTag;
     }
 
-    public void setSessionTopicWxappVideoTag(String sessionTopicWxappVideoTag) {
-        this.sessionTopicWxappVideoTag = sessionTopicWxappVideoTag;
+    public void setTaskTopicExamStudentTag(String taskTopicExamStudentTag) {
+        this.taskTopicExamStudentTag = taskTopicExamStudentTag;
+    }
+
+    public String getTaskConsumerExamStudentGroup() {
+        return taskConsumerExamStudentGroup;
+    }
+
+    public void setTaskConsumerExamStudentGroup(String taskConsumerExamStudentGroup) {
+        this.taskConsumerExamStudentGroup = taskConsumerExamStudentGroup;
     }
 
     public String getSessionTopicWxappAnswerTag() {
@@ -46,12 +84,20 @@ public class MqConfigDomain {
         this.sessionTopicWxappAnswerTag = sessionTopicWxappAnswerTag;
     }
 
-    public String getSessionConsumerWxappVideoGroup() {
-        return sessionConsumerWxappVideoGroup;
+    public String getSessionTopicWxappMonitorTag() {
+        return sessionTopicWxappMonitorTag;
+    }
+
+    public void setSessionTopicWxappMonitorTag(String sessionTopicWxappMonitorTag) {
+        this.sessionTopicWxappMonitorTag = sessionTopicWxappMonitorTag;
+    }
+
+    public String getSessionConsumerWxappMonitorGroup() {
+        return sessionConsumerWxappMonitorGroup;
     }
 
-    public void setSessionConsumerWxappVideoGroup(String sessionConsumerWxappVideoGroup) {
-        this.sessionConsumerWxappVideoGroup = sessionConsumerWxappVideoGroup;
+    public void setSessionConsumerWxappMonitorGroup(String sessionConsumerWxappMonitorGroup) {
+        this.sessionConsumerWxappMonitorGroup = sessionConsumerWxappMonitorGroup;
     }
 
     public String getSessionConsumerWxappAnswerGroup() {

themis-business/src/main/java/com/qmth/themis/business/entity/TBExamInvigilateUser.java

@@ -2,6 +2,7 @@ package com.qmth.themis.business.entity;
 
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
+import com.qmth.themis.common.contanst.Constants;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 
@@ -43,6 +44,7 @@ public class TBExamInvigilateUser implements Serializable {
     }
 
     public TBExamInvigilateUser(Long orgId, Long userId, String roomCode, String roomName) {
+        this.id = Constants.idGen.next();
         this.orgId = orgId;
         this.userId = userId;
         this.roomCode = roomCode;

themis-business/src/main/java/com/qmth/themis/business/entity/TBTaskHistory.java

@@ -3,6 +3,7 @@ package com.qmth.themis.business.entity;
 import com.baomidou.mybatisplus.annotation.FieldFill;
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
+import com.qmth.themis.common.contanst.Constants;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 
@@ -76,6 +77,23 @@ public class TBTaskHistory implements Serializable {
     @TableField(value = "finish_time")
     private Date finishTime;
 
+    public TBTaskHistory() {
+
+    }
+
+    public TBTaskHistory(Integer type, Long entityId, Integer status, String summary, Double progress, String importFileName, String importFilePath, Long createId) {
+        this.id = Constants.idGen.next();
+        this.type = type;
+        this.entityId = entityId;
+        this.status = status;
+        this.summary = summary;
+        this.progress = progress;
+        this.importFileName = importFileName;
+        this.importFilePath = importFilePath;
+        this.createId = createId;
+        this.startTime = new Date();
+    }
+
     public static long getSerialVersionUID() {
         return serialVersionUID;
     }

themis-business/src/main/java/com/qmth/themis/business/entity/TEStudent.java

@@ -2,6 +2,8 @@ package com.qmth.themis.business.entity;
 
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.qmth.themis.business.base.BaseEntity;
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.common.contanst.Constants;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 
@@ -51,6 +53,19 @@ public class TEStudent extends BaseEntity {
     @TableField(value = "enable")
     private Integer enable;
 
+    public TEStudent() {
+
+    }
+
+    public TEStudent(Long orgId, String identity, String name, Long createId) {
+        setId(Constants.idGen.next());
+        this.orgId = orgId;
+        this.identity = identity;
+        this.name = name;
+        this.password = SystemConstant.DEFAULT_PASSWORD;
+        setCreateId(createId);
+    }
+
     public Integer getEnable() {
         return enable;
     }

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

@@ -0,0 +1,31 @@
+package com.qmth.themis.business.enums;
+
+/**
+ * @Description: 考试模式 enum
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/16
+ */
+public enum ExamModeEnum {
+
+    /**
+     * 集中统一
+     */
+    together(0),
+
+    /**
+     * 随到随考
+     */
+    anytime(1);
+
+    private int id;
+
+    private ExamModeEnum(int id) {
+        this.id = id;
+    }
+
+    public int getId() {
+        return id;
+    }
+}

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

@@ -38,7 +38,12 @@ public enum MqEnum {
     /**
      * 用户轨迹
      */
-    USER_LOG(6, "用户轨迹");
+    USER_LOG(6, "用户轨迹"),
+
+    /**
+     * 任务
+     */
+    TASK_LOG(7, "任务");
 
     private int id;
 
@@ -66,6 +71,8 @@ public enum MqEnum {
             return EXCEPTION_LOG.getId();
         } else if (Objects.equals(value.trim(), USER_LOG.name())) {
             return USER_LOG.getId();
+        } else if (Objects.equals(value.trim(), TASK_LOG.name())) {
+            return TASK_LOG.getId();
         } else {
             return MESSAGE_LOG.getId();
         }
@@ -88,6 +95,8 @@ public enum MqEnum {
             return EXCEPTION_LOG.name();
         } else if (Objects.equals(value.trim(), USER_LOG.getCode())) {
             return USER_LOG.name();
+        } else if (Objects.equals(value.trim(), TASK_LOG.getCode())) {
+            return TASK_LOG.name();
         } else {
             return MESSAGE_LOG.name();
         }

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

@@ -21,7 +21,11 @@ public enum RoleEnum {
     /**
      * 考务老师
      */
-    STUDENT(3, "学生");
+    STUDENT(3, "学生"),
+    /**
+     * 监考员
+     */
+    INVIGILATE(4, "监考员");
 
     private int id;
 
@@ -43,6 +47,8 @@ public enum RoleEnum {
             return SUPER_ADMIN.getId();
         } else if (Objects.equals(value.trim(), TEACHER.name())) {
             return TEACHER.getId();
+        } else if (Objects.equals(value.trim(), TEACHER.name())) {
+            return INVIGILATE.getId();
         } else {
             return STUDENT.getId();
         }
@@ -59,6 +65,8 @@ public enum RoleEnum {
             return SUPER_ADMIN.name();
         } else if (Objects.equals(value.trim(), TEACHER.getCode())) {
             return TEACHER.name();
+        } else if (Objects.equals(value.trim(), TEACHER.getCode())) {
+            return INVIGILATE.name();
         } else {
             return STUDENT.name();
         }

themis-business/src/main/java/com/qmth/themis/business/forkjoin/ForkJoinTask.java

@@ -1,64 +1,105 @@
 package com.qmth.themis.business.forkjoin;
 
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.google.gson.Gson;
+import com.qmth.themis.business.constant.SpringContextHolder;
 import com.qmth.themis.business.dto.ExamStudentDtoImport;
-import com.qmth.themis.business.entity.TBTaskHistory;
-import com.qmth.themis.business.entity.TEExamStudent;
+import com.qmth.themis.business.entity.*;
+import com.qmth.themis.business.enums.RoleEnum;
 import com.qmth.themis.business.enums.TaskTypeExecEnum;
 import com.qmth.themis.business.service.TBTaskHistoryService;
+import com.qmth.themis.business.service.TBUserRoleService;
 import com.qmth.themis.business.service.TEExamStudentService;
+import com.qmth.themis.business.service.TEStudentService;
+import com.qmth.themis.common.exception.BusinessException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 
-import javax.annotation.Resource;
 import java.math.BigDecimal;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
+import java.util.*;
 import java.util.concurrent.RecursiveTask;
 
-@Component
 public class ForkJoinTask extends RecursiveTask<List<TEExamStudent>> {
     private final static Logger log = LoggerFactory.getLogger(ForkJoinTask.class);
 
     private int threshold = 500, start, end;//阀值，开始，结束，类型
     private List<Object> examStudentDtoImportList;
-    private TBTaskHistory tbTaskHistory = null;
+    private Map<String, Object> map = null;
 
     public ForkJoinTask() {
 
     }
 
-    public ForkJoinTask(int threshold, int start, int end, List<Object> examStudentDtoImportList, TBTaskHistory tbTaskHistory) {
+    public ForkJoinTask(int threshold, int start, int end, List<Object> examStudentDtoImportList, Map<String, Object> map) {
         this.threshold = threshold;
         this.start = start;
         this.end = end;
         this.examStudentDtoImportList = examStudentDtoImportList;
-        this.tbTaskHistory = tbTaskHistory;
+        this.map = map;
     }
 
-    @Resource
-    TEExamStudentService teExamStudentService;
-
-    @Resource
-    TBTaskHistoryService tbTaskHistoryService;
-
     @Override
+    @Transactional
     protected List<TEExamStudent> compute() {
         Gson gson = new Gson();
         List<TEExamStudent> teExamStudentList = new ArrayList<>();
+        Map tbTaskHistoryMap = (Map) map.get("tbTaskHistory");
+        tbTaskHistoryMap = timeTransform(tbTaskHistoryMap);
+        TBTaskHistory tbTaskHistory = gson.fromJson(gson.toJson(tbTaskHistoryMap), TBTaskHistory.class);
+        Long examId = Long.parseLong(String.valueOf(map.get("examId")));
+        Long orgId = Long.parseLong(String.valueOf(map.get("orgId")));
+        Long createId = Long.parseLong(String.valueOf(map.get("createId")));
+        Map<String, TEExamActivity> teExamActivityMap = (Map<String, TEExamActivity>) map.get("teExamActivityMap");
         if ((end - start) <= threshold) {
+            TEExamStudentService teExamStudentService = SpringContextHolder.getBean(TEExamStudentService.class);
+            TBTaskHistoryService tbTaskHistoryService = SpringContextHolder.getBean(TBTaskHistoryService.class);
+            TEStudentService teStudentService = SpringContextHolder.getBean(TEStudentService.class);
+            TBUserRoleService tbUserRoleService = SpringContextHolder.getBean(TBUserRoleService.class);
             for (int i = start; i <= end; i++) {
                 ExamStudentDtoImport examStudentDtoImport = (ExamStudentDtoImport) examStudentDtoImportList.get(i);
-                TEExamStudent teExamStudent = gson.fromJson(gson.toJson(examStudentDtoImport), TEExamStudent.class);
+                Map m = (Map) teExamActivityMap.get(examStudentDtoImport.getExamActivityCode());
+                m = timeTransform(m);
+                TEExamActivity teExamActivity = gson.fromJson(gson.toJson(m), TEExamActivity.class);
+                if (Objects.isNull(teExamActivity)) {
+                    throw new BusinessException("没有" + examStudentDtoImport.getExamActivityCode() + "的考场");
+                }
+                //先根据证件号+科目代码查询考生是否存在，存在则更新，不存在则插入
+                QueryWrapper<TEExamStudent> teExamStudentQueryWrapper = new QueryWrapper<>();
+                teExamStudentQueryWrapper.lambda().eq(TEExamStudent::getIdentity, examStudentDtoImport.getIdentity()).eq(TEExamStudent::getCourseCode, examStudentDtoImport.getCourseCode());
+                TEExamStudent teExamStudent = teExamStudentService.getOne(teExamStudentQueryWrapper);
+                //如果为空则插入考生数据，插入考生前先插入学生档案数据
+                if (Objects.isNull(teExamStudent)) {
+                    //先插入学生档案数据
+                    TEStudent teStudent = new TEStudent(orgId, examStudentDtoImport.getIdentity(), examStudentDtoImport.getName(), createId);
+                    teStudentService.save(teStudent);
+
+                    //插入用户角色关系
+                    TBUserRole tbUserRole = new TBUserRole(teStudent.getId(), RoleEnum.STUDENT.name());
+                    tbUserRoleService.save(tbUserRole);
+
+                    teExamStudent = gson.fromJson(gson.toJson(examStudentDtoImport), TEExamStudent.class);
+                    teExamStudent.setExamId(examId);
+                    teExamStudent.setExamActivityId(teExamActivity.getId());
+                    teExamStudent.setStudentId(teStudent.getId());
+                    teExamStudent.setCreateId(createId);
+                } else {
+                    teExamStudent.setUpdateId(createId);
+                    teExamStudent.setName(examStudentDtoImport.getName());
+                    teExamStudent.setCourseName(examStudentDtoImport.getCourseName());
+                    teExamStudent.setGrade(examStudentDtoImport.getGrade());
+                    teExamStudent.setClassNo(examStudentDtoImport.getClassNo());
+                    teExamStudent.setRoomCode(examStudentDtoImport.getRoomCode());
+                    teExamStudent.setRoomName(examStudentDtoImport.getRoomName());
+                    teExamStudent.setExamActivityId(teExamActivity.getId());
+                }
                 teExamStudentList.add(teExamStudent);
             }
             BigDecimal bigDecimal = new BigDecimal(100);
             int size = examStudentDtoImportList.size();
-            BigDecimal endBigDecimal = new BigDecimal(end);
+            BigDecimal endBigDecimal = new BigDecimal(teExamStudentList.size());
             BigDecimal progress = new BigDecimal(Double.valueOf(endBigDecimal.divide(new BigDecimal(size), 2, BigDecimal.ROUND_HALF_UP).multiply(bigDecimal).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue())).setScale(0, BigDecimal.ROUND_HALF_UP);
-            teExamStudentService.saveBatch(teExamStudentList);
+            teExamStudentService.saveOrUpdateBatch(teExamStudentList);
             if (progress.intValue() == 100) {
                 tbTaskHistory.setStatus(TaskTypeExecEnum.finish.ordinal());
                 tbTaskHistory.setSummary("共处理了" + size + "条数据");
@@ -73,12 +114,42 @@ public class ForkJoinTask extends RecursiveTask<List<TEExamStudent>> {
         } else {
             // 将大任务分解成两个小任务
             int middle = (start + end) / 2;
-            ForkJoinTask left = new ForkJoinTask(threshold, start, middle, examStudentDtoImportList, tbTaskHistory);
-            ForkJoinTask right = new ForkJoinTask(threshold, middle + 1, end, examStudentDtoImportList, tbTaskHistory);
+            ForkJoinTask left = new ForkJoinTask(threshold, start, middle, examStudentDtoImportList, map);
+            ForkJoinTask right = new ForkJoinTask(threshold, middle + 1, end, examStudentDtoImportList, map);
             invokeAll(left, right);
             teExamStudentList.addAll(left.join());
             teExamStudentList.addAll(right.join());
         }
         return teExamStudentList;
     }
+
+    /**
+     * 毫秒时间转换
+     *
+     * @param map
+     * @return
+     */
+    Map timeTransform(Map map) {
+        if (Objects.nonNull(map.get("createTime")) && map.get("createTime") instanceof Long) {
+            Date date = new Date();
+            date.setTime(Long.parseLong(String.valueOf(map.get("createTime"))));
+            map.put("createTime", date);
+        }
+        if (Objects.nonNull(map.get("startTime")) && map.get("startTime") instanceof Long) {
+            Date date = new Date();
+            date.setTime(Long.parseLong(String.valueOf(map.get("startTime"))));
+            map.put("startTime", date);
+        }
+        if (Objects.nonNull(map.get("finishTime")) && map.get("finishTime") instanceof Long) {
+            Date date = new Date();
+            date.setTime(Long.parseLong(String.valueOf(map.get("finishTime"))));
+            map.put("finishTime", date);
+        }
+        if (Objects.nonNull(map.get("updateTime")) && map.get("updateTime") instanceof Long) {
+            Date date = new Date();
+            date.setTime(Long.parseLong(String.valueOf(map.get("updateTime"))));
+            map.put("updateTime", date);
+        }
+        return map;
+    }
 }

themis-business/src/main/java/com/qmth/themis/business/service/impl/TBAttachmentServiceImpl.java

@@ -8,6 +8,8 @@ import com.qmth.themis.business.entity.TBAttachment;
 import com.qmth.themis.business.enums.UploadFileEnum;
 import com.qmth.themis.business.service.TBAttachmentService;
 import com.qmth.themis.business.util.OssUtil;
+import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.common.enums.ExceptionResultEnum;
 import com.qmth.themis.common.exception.BusinessException;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.io.FileUtils;
@@ -86,6 +88,7 @@ public class TBAttachmentServiceImpl extends ServiceImpl<TBAttachmentMapper, TBA
                 throw new BusinessException("md5不一致");
             }
             tbAttachment = new TBAttachment();
+            tbAttachment.setId(Constants.idGen.next());
             tbAttachment.setPath(path);
             tbAttachment.setName(fileName);
             tbAttachment.setType(format);
@@ -97,7 +100,7 @@ public class TBAttachmentServiceImpl extends ServiceImpl<TBAttachmentMapper, TBA
             StringJoiner stringJoiner = new StringJoiner("");
             String uploadType = String.valueOf(map.get(SystemConstant.UPLOAD_TYPE));
             if (Objects.isNull(uploadType)) {
-                throw new BusinessException("请上传文件类型");
+                throw new BusinessException(ExceptionResultEnum.ATTACHMENT_TYPE_IS_NULL);
             } else if (Objects.equals(uploadType, UploadFileEnum.base_photo.name())) {
                 stringJoiner.add(uploadType)
                         .add(File.separator).add(String.valueOf(orgId));
@@ -127,6 +130,7 @@ public class TBAttachmentServiceImpl extends ServiceImpl<TBAttachmentMapper, TBA
                 jsonObject.put(SystemConstant.TYPE, SystemConstant.LOCAL);
             }
             jsonObject.put(SystemConstant.PATH, stringJoiner.toString());
+            jsonObject.put(SystemConstant.UPLOAD_TYPE, UploadFileEnum.valueOf(uploadType).ordinal());
             tbAttachment.setRemark(jsonObject.toJSONString());
             tbAttachment.setCreateId(userId);
             this.save(tbAttachment);

themis-business/src/main/java/com/qmth/themis/business/templete/TaskImportTemplete.java

@@ -1,10 +1,11 @@
 package com.qmth.themis.business.templete;
 
-import com.qmth.themis.business.entity.TBTaskHistory;
 import com.qmth.themis.common.util.Result;
 import org.springframework.web.multipart.MultipartFile;
 
+import java.io.File;
 import java.io.IOException;
+import java.util.Map;
 
 /**
  * @Description: 导入任务模版
@@ -15,5 +16,5 @@ import java.io.IOException;
  */
 public interface TaskImportTemplete {
 
-    Result importTask(MultipartFile file, TBTaskHistory tbTaskHistory) throws IOException;
+    Result importTask(File file, Map<String, Object> map) throws IOException, InterruptedException;
 }

themis-business/src/main/java/com/qmth/themis/business/templete/impl/ExamStudentTemplete.java

@@ -2,10 +2,13 @@ package com.qmth.themis.business.templete.impl;
 
 import com.alibaba.fastjson.JSONObject;
 import com.google.common.collect.Lists;
+import com.qmth.themis.business.constant.SpringContextHolder;
 import com.qmth.themis.business.dto.ExamStudentDtoImport;
-import com.qmth.themis.business.entity.TBTaskHistory;
+import com.qmth.themis.business.entity.TBExamInvigilateUser;
 import com.qmth.themis.business.entity.TEExamStudent;
+import com.qmth.themis.business.enums.ExamModeEnum;
 import com.qmth.themis.business.forkjoin.ForkJoinTask;
+import com.qmth.themis.business.service.TBExamInvigilateUserService;
 import com.qmth.themis.business.templete.TaskImportTemplete;
 import com.qmth.themis.business.util.ExcelCallback;
 import com.qmth.themis.business.util.ExcelError;
@@ -16,8 +19,9 @@ import com.qmth.themis.common.util.ResultUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.LinkedMultiValueMap;
-import org.springframework.web.multipart.MultipartFile;
 
+import java.io.File;
+import java.io.FileInputStream;
 import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.ExecutionException;
@@ -35,8 +39,12 @@ public class ExamStudentTemplete implements TaskImportTemplete {
     private final static Logger log = LoggerFactory.getLogger(ExamStudentTemplete.class);
 
     @Override
-    public Result importTask(MultipartFile file, TBTaskHistory tbTaskHistory) throws IOException {
-        List<LinkedMultiValueMap<Integer, Object>> finalList = ExcelUtil.excelReader(file.getInputStream(), Lists.newArrayList(ExamStudentDtoImport.class), new ExcelCallback() {
+    public Result importTask(File file, Map<String, Object> map) throws IOException, InterruptedException {
+        Integer mode = Integer.parseInt(String.valueOf(map.get("mode")));
+        Long orgId = Long.parseLong(String.valueOf(map.get("orgId")));
+        Long createId = Long.parseLong(String.valueOf(map.get("createId")));
+        FileInputStream inputStream = new FileInputStream(file);
+        List<LinkedMultiValueMap<Integer, Object>> finalList = ExcelUtil.excelReader(inputStream, Lists.newArrayList(ExamStudentDtoImport.class), new ExcelCallback() {
             @Override
             public List<LinkedMultiValueMap<Integer, Object>> callback(List<LinkedMultiValueMap<Integer, Object>> finalList, List<LinkedMultiValueMap<Integer, String>> finalColumnNameList) throws IllegalAccessException {
                 List<ExcelError> excelErrorList = new ArrayList<>();
@@ -50,6 +58,10 @@ public class ExamStudentTemplete implements TaskImportTemplete {
                         if (excelErrorTemp.size() > 0) {
                             excelErrorList.addAll(excelErrorTemp);
                         }
+                        //校验考试模式，如果是集中统一，则需填写考试场次
+                        if (mode.intValue() == ExamModeEnum.together.ordinal() && Objects.isNull(examStudentDtoImport.getExamActivityCode())) {
+                            excelErrorList.add(new ExcelError(y + 1, "excel第" + (i + 1) + "个sheet第" + (y + 1) + "行考场为空"));
+                        }
                     }
                 }
                 if (excelErrorList.size() > 0) {
@@ -62,26 +74,49 @@ public class ExamStudentTemplete implements TaskImportTemplete {
         if (Objects.nonNull(finalList) && finalList.size() > 0) {
             log.info("开始forkjoin数据拆分");
             long start = System.currentTimeMillis();
+            ForkJoinPool forkJoinPool = new ForkJoinPool();
             for (int i = 0; i < finalList.size(); i++) {
-                LinkedMultiValueMap<Integer, Object> map = finalList.get(i);
-                List<Object> examStudentDtoImportList = map.get(i);
-                ForkJoinTask forkJoinTask = new ForkJoinTask(500, 0, examStudentDtoImportList.size() - 1, examStudentDtoImportList, tbTaskHistory);
-                ForkJoinPool forkJoinPool = new ForkJoinPool();
+                LinkedMultiValueMap<Integer, Object> multiValueMap = finalList.get(i);
+                List<Object> examStudentDtoImportList = multiValueMap.get(i);
+                ForkJoinTask forkJoinTask = new ForkJoinTask(500, 0, examStudentDtoImportList.size() - 1, examStudentDtoImportList, map);
                 Future<List<TEExamStudent>> result = forkJoinPool.submit(forkJoinTask);
-                if (forkJoinTask.isCompletedAbnormally()) { // 来检查任务是否已经抛出异常或已经被取消了，要注意此方法。由于提交任务之后，检测该任务是否有异常，不是阻塞的。所以需要上面的等待任务的完成。才能正确的获取到是否有异常
+                if (forkJoinTask.isCompletedAbnormally()) {
+                    //检查任务是否已经抛出异常或已经被取消了，要注意此方法。由于提交任务之后，检测该任务是否有异常，不是阻塞的。所以需要上面的等待任务的完成。才能正确的获取到是否有异常
                     throw new BusinessException(forkJoinTask.getException().getMessage());
                 }
                 try {
                     List<TEExamStudent> list = result.get();
+                    Map<String, String> tbExamInvigilateUserMap = (Map<String, String>) map.get("tbExamInvigilateUserMap");
+                    if (Objects.nonNull(list) && list.size() > 0) {
+                        //考场的创建
+                        TBExamInvigilateUserService tbExamInvigilateUserService = SpringContextHolder.getBean(TBExamInvigilateUserService.class);
+                        Set<String> roomCodeAndNameSet = new HashSet<>();
+                        list.forEach(s -> {
+                            roomCodeAndNameSet.add(s.getRoomCode() + ":" + s.getRoomName());
+                        });
+                        List<TBExamInvigilateUser> tbExamInvigilateUserList = new ArrayList<>();
+                        roomCodeAndNameSet.forEach(s -> {
+                            if (Objects.isNull(tbExamInvigilateUserMap) || (Objects.nonNull(tbExamInvigilateUserMap) && Objects.isNull(tbExamInvigilateUserMap.get(s)))) {
+                                String[] strs = s.split(":");
+                                TBExamInvigilateUser tbExamInvigilateUser = new TBExamInvigilateUser(orgId, createId, strs[0], strs[1]);
+                                tbExamInvigilateUserList.add(tbExamInvigilateUser);
+                            }
+                        });
+                        tbExamInvigilateUserService.saveBatch(tbExamInvigilateUserList);
+                    }
                     log.info("list.size:{}", list.size());
                 } catch (ExecutionException | InterruptedException e) {
                     e.printStackTrace();
+                } finally {
+                    if (Objects.nonNull(forkJoinPool)) {
+                        forkJoinPool.shutdown();
+                    }
                 }
             }
             long end = System.currentTimeMillis();
             log.info("forkjoin数据拆分结束,============耗时============:{}秒", (end - start) / 1000);
         }
-        Map map = new HashMap<>();
-        return ResultUtil.ok(map);
+        Map resultMap = new HashMap<>();
+        return ResultUtil.ok(resultMap);
     }
 }

themis-business/src/main/java/com/qmth/themis/business/util/RedisUtil.java

@@ -1,12 +1,14 @@
 package com.qmth.themis.business.util;
 
 import com.qmth.themis.business.constant.SystemConstant;
+import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
@@ -179,7 +181,7 @@ public class RedisUtil {
      * @param hashKey
      * @param hashValue
      */
-    public void setSessionTopicList(String key, String hashKey, Object hashValue) {
+    public void setMqTopicList(String key, String hashKey, Object hashValue) {
         redisTemplate.opsForHash().put(key, hashKey, hashValue);
     }
 
@@ -190,7 +192,7 @@ public class RedisUtil {
      * @param hashKey
      * @return
      */
-    public Object getSessionTopicList(String key, String hashKey) {
+    public Object getMqTopicList(String key, String hashKey) {
         return redisTemplate.opsForHash().get(key, hashKey);
     }
 
@@ -200,7 +202,7 @@ public class RedisUtil {
      * @param key
      * @param hashKey
      */
-    public void deleteSessionTopicList(String key, String hashKey) {
+    public void deleteMqTopicList(String key, String hashKey) {
         redisTemplate.opsForHash().delete(key, hashKey);
     }
 
@@ -223,4 +225,56 @@ public class RedisUtil {
     public Map getHashEntries(String key) {
         return redisTemplate.opsForHash().entries(key);
     }
+
+    /**
+     * 分布式锁
+     *
+     * @param key
+     * @param hashKey
+     * @param time
+     * @param timeUnit
+     * @return
+     */
+    public boolean lock(String key, String hashKey, long time, TimeUnit timeUnit) {
+        String lock = key + hashKey;
+        // 利用lambda表达式
+        return (Boolean) redisTemplate.execute((RedisCallback) connection -> {
+            long expireAt = System.currentTimeMillis() + time + 1;
+            Boolean acquire = connection.setNX(lock.getBytes(), String.valueOf(expireAt).getBytes());
+            if (acquire) {
+                redisTemplate.expire(lock, time, timeUnit);
+                return true;
+            } else {
+                byte[] value = connection.get(lock.getBytes());
+                if (Objects.nonNull(value) && value.length > 0) {
+                    long expireTime = Long.parseLong(new String(value));
+                    // 如果锁已经过期
+                    if (expireTime < System.currentTimeMillis()) {
+                        // 重新加锁，防止死锁
+                        byte[] oldValue = connection.getSet(lock.getBytes(), String.valueOf(System.currentTimeMillis() + time + 1).getBytes());
+                        redisTemplate.expire(lock, time, timeUnit);
+                        return Long.parseLong(new String(oldValue)) < System.currentTimeMillis();
+                    }
+                }
+            }
+            return false;
+        });
+    }
+
+    /**
+     * 删除锁
+     *
+     * @param key
+     * @param hashKey
+     * @return
+     */
+    public boolean releaseLock(String key, String hashKey) {
+        Object result = redisTemplate.opsForValue().get(key + hashKey);
+        if (Objects.nonNull(result)) {
+            redisTemplate.delete(key + hashKey);
+            return true;
+        } else {
+            return false;
+        }
+    }
 }

themis-common/src/main/java/com/qmth/themis/common/enums/ExceptionResultEnum.java

@@ -64,7 +64,9 @@ public enum ExceptionResultEnum {
 
     USER_NO("102", "用户不存在"),
 
-    EXAM_NO("102", "考试批次不存在"),
+    EXAM_NO("102", "考试批次信息不存在"),
+
+    EXAM_ACTIVITY_NO("102", "考试场次信息不存在"),
 
     TOKEN_NO("106", "token已过期"),
 

themis-common/src/main/java/com/qmth/themis/common/enums/Platform.java

@@ -2,7 +2,7 @@ package com.qmth.themis.common.enums;
 
 public enum Platform {
 
-    web("web"), wap("web"), wxapp_video("wxapp_video"), wxapp_answer("wxapp_answer"), win("pc");
+    web("web"), wap("web"), wxapp_monitor("wxapp_monitor"), wxapp_answer("wxapp_answer"), win("pc");
 
     public static Platform findByName(String name) {
         if (name == null) {

themis-common/src/main/java/com/qmth/themis/common/enums/Source.java

@@ -2,7 +2,7 @@ package com.qmth.themis.common.enums;
 
 public enum Source {
 
-    web, wxapp_video, wxapp_answer, pc;
+    web, wxapp_monitor, wxapp_answer, pc;
 
     public static Source findByName(String name) {
         if (name == null) {

themis-mq/src/main/java/com/qmth/themis/mq/listener/RocketSessionConsumer.java

@@ -1,217 +1,160 @@
-package com.qmth.themis.mq.listener;
-
-import com.google.gson.Gson;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.entity.TBSession;
-import com.qmth.themis.business.entity.TMRocketMessage;
-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.mq.dto.MqDto;
-import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
-import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
-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.List;
-import java.util.Objects;
-
-/**
- * @Description: 普通消息监听 session_topic
- * @Param:
- * @return:
- * @Author: wangliang
- * @Date: 2020/6/28
- */
-@Service
-public class RocketSessionConsumer implements
-//        MessageListenerOrderly
-        MessageListenerConcurrently //并发消费
-{
-
-    private final static Logger log = LoggerFactory.getLogger(RocketSessionConsumer.class);
-
-    @Resource
-    TBSessionService tbSessionService;
-
-    @Resource
-    RedisUtil redisUtil;
-
-    @Resource
-    TMRocketMessageService tmRocketMessageService;
-
-    /**
-     * 并发消费
-     *
-     * @param msgs
-     * @param consumeConcurrentlyContext
-     * @return
-     */
-    @Override
-    @Transactional
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            Gson gson = new Gson();
-            for (MessageExt messageExt : msgs) {
-                log.info(":{}-:{} sessionConsumer 重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
-                MqDto mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.info(":{}-:{} sessionConsumer 接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
-                log.info(":{}-:{} sessionConsumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
-                if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
-                    log.info(":{}-:{} 更新db", threadId, threadName);
-                    tbSessionService.saveSessionInfo(JacksonUtil.readJson(JacksonUtil.parseJson(mqDto.getBody()), TBSession.class), mqDto.getTimestamp());
-                    mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
-                    TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
-                    tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
-                    tmRocketMessageService.save(tmRocketMessage);
-                    redisUtil.deleteSessionTopicList(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId());
-                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                } else {
-                    log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-                }
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
-    }
-
+//package com.qmth.themis.mq.listener;
+//
+//import com.google.gson.Gson;
+//import com.qmth.themis.business.constant.SystemConstant;
+//import com.qmth.themis.business.entity.TBSession;
+//import com.qmth.themis.business.entity.TMRocketMessage;
+//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.mq.dto.MqDto;
+//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+//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.List;
+//import java.util.Objects;
+//
+///**
+// * @Description: 普通消息监听 session_topic
+// * @Param:
+// * @return:
+// * @Author: wangliang
+// * @Date: 2020/6/28
+// */
+//@Service
+//public class RocketSessionConsumer implements
+////        MessageListenerOrderly
+//        MessageListenerConcurrently //并发消费
+//{
+//
+//    private final static Logger log = LoggerFactory.getLogger(RocketSessionConsumer.class);
+//
+//    @Resource
+//    TBSessionService tbSessionService;
+//
+//    @Resource
+//    RedisUtil redisUtil;
+//
+//    @Resource
+//    TMRocketMessageService tmRocketMessageService;
+//
 //    /**
-//     * 顺序消费
+//     * 并发消费
 //     *
 //     * @param msgs
-//     * @param consumeOrderlyContext
+//     * @param consumeConcurrentlyContext
 //     * @return
 //     */
 //    @Override
-//    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext
-//            consumeOrderlyContext) {
+//    @Transactional
+//    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
 //        try {
+//            long threadId = Thread.currentThread().getId();
+//            String threadName = Thread.currentThread().getName();
+//            Gson gson = new Gson();
 //            for (MessageExt messageExt : msgs) {
-//                log.info("sessionConsumer重试次数:{}", messageExt.getReconsumeTimes());
-//                MqDto mqDto = (MqDto) toJavaObject(parseObject(new String(messageExt.getBody(), Constants.CHARSET)), MqDto.class);
-//                log.info("sessionConsumer接受到的消息:{}", JacksonUtil.parseJson(mqDto));
-//                log.info("mqDto sequence:{},tag:{}", mqDto.getSequence(), mqDto.getTag());
-//                MqDto redisMqdto = (MqDto) redisUtil.getSessionTopicList(mqDto.getId());
-//                if (Objects.nonNull(redisMqdto)) {
-//                    if (Objects.nonNull(redisMqdto.getAck()) && redisMqdto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
-//                        log.info("更新db");
-//                        tbSessionService.saveSessionInfo(toJavaObject((JSON) mqDto.getBody(), TBSession.class), redisMqdto.getTimestamp());
-//                        redisUtil.deleteSessionTopicList(redisMqdto.getId());
-//                        return ConsumeOrderlyStatus.SUCCESS;
-//                    } else {
-//                        log.info("消息ack未确认，重发");
-//                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
-//                    }
+//                log.info(":{}-:{} sessionConsumer 重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+//                MqDto mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+//                log.info(":{}-:{} sessionConsumer 接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
+//                log.info(":{}-:{} sessionConsumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
+//                if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
+//                    log.info(":{}-:{} 更新db", threadId, threadName);
+//                    tbSessionService.saveSessionInfo(JacksonUtil.readJson(JacksonUtil.parseJson(mqDto.getBody()), TBSession.class), mqDto.getTimestamp());
+//                    mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
+//                    TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+//                    tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
+//                    tmRocketMessageService.save(tmRocketMessage);
+//                    redisUtil.deleteSessionTopicList(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId());
+//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
 //                } else {
-//                    log.info("消息数据为空，重发消息");
-//                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+//                    log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
+//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
 //                }
 //            }
 //        } catch (Exception e) {
 //            e.printStackTrace();
-//            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+//            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+//        }
+//        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+//    }
+//
+////    /**
+////     * 顺序消费
+////     *
+////     * @param msgs
+////     * @param consumeOrderlyContext
+////     * @return
+////     */
+////    @Override
+////    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext
+////            consumeOrderlyContext) {
+////        try {
+////            for (MessageExt messageExt : msgs) {
+////                log.info("sessionConsumer重试次数:{}", messageExt.getReconsumeTimes());
+////                MqDto mqDto = (MqDto) toJavaObject(parseObject(new String(messageExt.getBody(), Constants.CHARSET)), MqDto.class);
+////                log.info("sessionConsumer接受到的消息:{}", JacksonUtil.parseJson(mqDto));
+////                log.info("mqDto sequence:{},tag:{}", mqDto.getSequence(), mqDto.getTag());
+////                MqDto redisMqdto = (MqDto) redisUtil.getSessionTopicList(mqDto.getId());
+////                if (Objects.nonNull(redisMqdto)) {
+////                    if (Objects.nonNull(redisMqdto.getAck()) && redisMqdto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
+////                        log.info("更新db");
+////                        tbSessionService.saveSessionInfo(toJavaObject((JSON) mqDto.getBody(), TBSession.class), redisMqdto.getTimestamp());
+////                        redisUtil.deleteSessionTopicList(redisMqdto.getId());
+////                        return ConsumeOrderlyStatus.SUCCESS;
+////                    } else {
+////                        log.info("消息ack未确认，重发");
+////                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+////                    }
+////                } else {
+////                    log.info("消息数据为空，重发消息");
+////                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+////                }
+////            }
+////        } catch (Exception e) {
+////            e.printStackTrace();
+////            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+////        }
+////        return ConsumeOrderlyStatus.SUCCESS;//成功
+////    }
+//
+//    @Service
+//    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWebGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWebTag}")
+//    public class sessionConsumerWeb 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.setMessageModel(MessageModel.BROADCASTING);
+//            defaultMQPushConsumer.registerMessageListener(RocketSessionConsumer.this::consumeMessage);
 //        }
-//        return ConsumeOrderlyStatus.SUCCESS;//成功
 //    }
-
-    @Service
-    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWebGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWebTag}")
-    public class sessionConsumerWeb 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.setMessageModel(MessageModel.BROADCASTING);
-            defaultMQPushConsumer.registerMessageListener(RocketSessionConsumer.this::consumeMessage);
-        }
-    }
-
-    @Service
-    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerPcGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicPcTag}")
-    public class sessionConsumerPc 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(RocketSessionConsumer.this::consumeMessage);
-        }
-    }
-
-    @Service
-    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWxappVideoGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWxappVideoTag}")
-    public class sessionConsumerWxappVideo 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(RocketSessionConsumer.this::consumeMessage);
-        }
-    }
-
-    @Service
-    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWxappAnswerGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWxappAnswerTag}")
-    public class sessionConsumerWxappAnswer 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(RocketSessionConsumer.this::consumeMessage);
-        }
-    }
-
-    /**
-     * 死信队列
-     */
+//
 //    @Service
-//    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerGroupDlq}", topic = "${mq.config.sessionTopicDlq}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicTag}")
-//    public class dlqSessionConsumer implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+//    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerPcGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicPcTag}")
+//    public class sessionConsumerPc implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
 //
 //        @Override
 //        public void onMessage(Message message) {
@@ -220,10 +163,67 @@ public class RocketSessionConsumer implements
 //
 //        @Override
 //        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
-//            log.info("dlqSessionConsumer死信队列进来了");
 //            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
+//            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 //            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
-//            defaultMQPushConsumer.registerMessageListener(RocketConsumer.this::consumeMessage);
+//            defaultMQPushConsumer.registerMessageListener(RocketSessionConsumer.this::consumeMessage);
 //        }
 //    }
-}
+//
+//    @Service
+//    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWxappMonitorGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWxappMonitorTag}")
+//    public class sessionConsumerWxappMonitor 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(RocketSessionConsumer.this::consumeMessage);
+//        }
+//    }
+//
+//    @Service
+//    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWxappAnswerGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWxappAnswerTag}")
+//    public class sessionConsumerWxappAnswer 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(RocketSessionConsumer.this::consumeMessage);
+//        }
+//    }
+//
+//    /**
+//     * 死信队列
+//     */
+////    @Service
+////    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerGroupDlq}", topic = "${mq.config.sessionTopicDlq}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicTag}")
+////    public class dlqSessionConsumer implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+////
+////        @Override
+////        public void onMessage(Message message) {
+////            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
+////        }
+////
+////        @Override
+////        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
+////            log.info("dlqSessionConsumer死信队列进来了");
+////            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
+////            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
+////            defaultMQPushConsumer.registerMessageListener(RocketConsumer.this::consumeMessage);
+////        }
+////    }
+//}

themis-mq/src/main/java/com/qmth/themis/mq/listener/RocketUserLogConsumer.java

@@ -1,122 +1,122 @@
-package com.qmth.themis.mq.listener;
-
-import com.google.gson.Gson;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.entity.TMRocketMessage;
-import com.qmth.themis.business.enums.MqEnum;
-import com.qmth.themis.business.enums.SystemOperationEnum;
-import com.qmth.themis.business.service.TEUserLogService;
-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.mq.dto.MqDto;
-import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
-import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
-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.List;
-import java.util.Objects;
-
-/**
- * @Description: 普通消息监听 用户日志
- * @Param:
- * @return:
- * @Author: wangliang
- * @Date: 2020/7/2
- */
-@Service
-public class RocketUserLogConsumer implements MessageListenerConcurrently {
-    private final static Logger log = LoggerFactory.getLogger(RocketUserLogConsumer.class);
-
-    @Resource
-    RedisUtil redisUtil;
-
-    @Resource
-    TEUserLogService teUserLogService;
-
-    @Resource
-    TMRocketMessageService tmRocketMessageService;
-
-    @Override
-    @Transactional
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            Gson gson = new Gson();
-            for (MessageExt messageExt : msgs) {
-                log.info(":{}-:{} userLogConsumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
-//                MqDto mqDto = (MqDto) toJavaObject(parseObject(new String(messageExt.getBody(), Constants.CHARSET)), MqDto.class);
-                MqDto mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.info(":{}-:{} userLogConsumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
-                log.info(":{}-:{} userLogConsumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
-                if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
-                    log.info(":{}-:{} 插入用户轨迹日志", threadId, threadName);
-                    teUserLogService.saveUserLogInfo(mqDto.getTimestamp(), mqDto.getObjId(), MqEnum.valueOf(String.valueOf(mqDto.getType())).getId(), SystemOperationEnum.valueOf(String.valueOf(mqDto.getBody())).getCode(), JacksonUtil.parseJson(mqDto));
-                    mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
-                    TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
-                    tmRocketMessageService.save(tmRocketMessage);
-                    redisUtil.deleteSessionTopicList(SystemConstant.USERLOG_TOPIC_BUFFER_LIST, mqDto.getId());
-                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                } else {
-                    log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-                }
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
-    }
-
-    @Service
-    @RocketMQMessageListener(consumerGroup = "${mq.config.userLogConsumerUserGroup}", topic = "${mq.config.userLogTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.userLogTopicUserTag}")
-    public class sessionConsumerUserLog 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(RocketUserLogConsumer.this::consumeMessage);
-        }
-    }
-
-    @Service
-    @RocketMQMessageListener(consumerGroup = "${mq.config.userLogConsumerStudentGroup}", topic = "${mq.config.userLogTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.userLogTopicStudentTag}")
-    public class sessionConsumerStudentLog 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(RocketUserLogConsumer.this::consumeMessage);
-        }
-    }
-}
+//package com.qmth.themis.mq.listener;
+//
+//import com.google.gson.Gson;
+//import com.qmth.themis.business.constant.SystemConstant;
+//import com.qmth.themis.business.entity.TMRocketMessage;
+//import com.qmth.themis.business.enums.MqEnum;
+//import com.qmth.themis.business.enums.SystemOperationEnum;
+//import com.qmth.themis.business.service.TEUserLogService;
+//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.mq.dto.MqDto;
+//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+//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.List;
+//import java.util.Objects;
+//
+///**
+// * @Description: 普通消息监听 用户日志
+// * @Param:
+// * @return:
+// * @Author: wangliang
+// * @Date: 2020/7/2
+// */
+//@Service
+//public class RocketUserLogConsumer implements MessageListenerConcurrently {
+//    private final static Logger log = LoggerFactory.getLogger(RocketUserLogConsumer.class);
+//
+//    @Resource
+//    RedisUtil redisUtil;
+//
+//    @Resource
+//    TEUserLogService teUserLogService;
+//
+//    @Resource
+//    TMRocketMessageService tmRocketMessageService;
+//
+//    @Override
+//    @Transactional
+//    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+//        try {
+//            long threadId = Thread.currentThread().getId();
+//            String threadName = Thread.currentThread().getName();
+//            Gson gson = new Gson();
+//            for (MessageExt messageExt : msgs) {
+//                log.info(":{}-:{} userLogConsumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+////                MqDto mqDto = (MqDto) toJavaObject(parseObject(new String(messageExt.getBody(), Constants.CHARSET)), MqDto.class);
+//                MqDto mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+//                log.info(":{}-:{} userLogConsumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
+//                log.info(":{}-:{} userLogConsumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
+//                if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
+//                    log.info(":{}-:{} 插入用户轨迹日志", threadId, threadName);
+//                    teUserLogService.saveUserLogInfo(mqDto.getTimestamp(), mqDto.getObjId(), MqEnum.valueOf(String.valueOf(mqDto.getType())).getId(), SystemOperationEnum.valueOf(String.valueOf(mqDto.getBody())).getCode(), JacksonUtil.parseJson(mqDto));
+//                    mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
+//                    TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+//                    tmRocketMessageService.save(tmRocketMessage);
+//                    redisUtil.deleteSessionTopicList(SystemConstant.USERLOG_TOPIC_BUFFER_LIST, mqDto.getId());
+//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+//                } else {
+//                    log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
+//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+//                }
+//            }
+//        } catch (Exception e) {
+//            e.printStackTrace();
+//            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+//        }
+//        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+//    }
+//
+//    @Service
+//    @RocketMQMessageListener(consumerGroup = "${mq.config.userLogConsumerUserGroup}", topic = "${mq.config.userLogTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.userLogTopicUserTag}")
+//    public class sessionConsumerUserLog 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(RocketUserLogConsumer.this::consumeMessage);
+//        }
+//    }
+//
+//    @Service
+//    @RocketMQMessageListener(consumerGroup = "${mq.config.userLogConsumerStudentGroup}", topic = "${mq.config.userLogTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.userLogTopicStudentTag}")
+//    public class sessionConsumerStudentLog 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(RocketUserLogConsumer.this::consumeMessage);
+//        }
+//    }
+//}

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

@@ -62,16 +62,7 @@ public class MqDtoServiceImpl implements MqDtoService {
                 mqDto.setAck(SystemConstant.UNSEND_ACK_TYPE);
             }
         } finally {
-            if (Objects.nonNull(mqDto)) {
-                switch (mqEnum.ordinal()) {
-                    case 0:
-                        redisUtil.setSessionTopicList(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
-                        break;
-                    default:
-                        redisUtil.setSessionTopicList(SystemConstant.USERLOG_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
-                        break;
-                }
-            }
+            setTopic(mqDto);
         }
         return mqDto;
     }
@@ -92,7 +83,25 @@ public class MqDtoServiceImpl implements MqDtoService {
             if (Objects.nonNull(mqDto)) {
                 mqDto.setAck(SystemConstant.UNSEND_ACK_TYPE);
             }
+        } finally {
+            setTopic(mqDto);
         }
         return null;
     }
+
+    void setTopic(MqDto mqDto) {
+        if (Objects.nonNull(mqDto)) {
+            switch (mqDto.getType().ordinal()) {
+                case 0:
+                    redisUtil.setMqTopicList(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
+                    break;
+                case 5:
+                    redisUtil.setMqTopicList(SystemConstant.USERLOG_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
+                    break;
+                default:
+                    redisUtil.setMqTopicList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
+                    break;
+            }
+        }
+    }
 }

themis-task/src/main/java/com/qmth/themis/task/config/DictionaryConfig.java

@@ -15,6 +15,17 @@ import org.springframework.context.annotation.Configuration;
 @Configuration
 public class DictionaryConfig {
 
+    /**
+     * 系统配置
+     *
+     * @return
+     */
+    @Bean
+    @ConfigurationProperties(prefix = "sys.config", ignoreUnknownFields = false)
+    public SysDomain sysDomain() {
+        return new SysDomain();
+    }
+
     /**
      * mq配置
      *
@@ -36,4 +47,15 @@ public class DictionaryConfig {
     public QuartzConfigDomain quartzConfigDomain() {
         return new QuartzConfigDomain();
     }
+
+    /**
+     * 阿里云oss配置
+     *
+     * @return
+     */
+    @Bean
+    @ConfigurationProperties(prefix = "aliyun.oss", ignoreUnknownFields = false)
+    public AliYunOssDomain aliYunOssDomain() {
+        return new AliYunOssDomain();
+    }
 }

themis-task/src/main/java/com/qmth/themis/task/constant/TaskConstant.java

@@ -0,0 +1,45 @@
+package com.qmth.themis.task.constant;
+
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.enums.UploadFileEnum;
+import com.qmth.themis.common.enums.ExceptionResultEnum;
+import com.qmth.themis.common.exception.BusinessException;
+import com.qmth.themis.task.config.DictionaryConfig;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * @Description: 后台系统常量
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/15
+ */
+public class TaskConstant {
+
+    /**
+     * 获取系统上传环境
+     *
+     * @param type
+     * @return
+     */
+    public static Map<String, Object> getAttachmentEnv(Integer type) {
+        if (Objects.isNull(type) || Objects.equals(type, "")) {
+            throw new BusinessException(ExceptionResultEnum.ATTACHMENT_TYPE_IS_NULL);
+        }
+        DictionaryConfig dictionaryConfig = SpringContextHolder.getBean(DictionaryConfig.class);
+        Map<String, Object> mapParameter = new HashMap<>();
+        mapParameter.put(SystemConstant.END_POINT, dictionaryConfig.aliYunOssDomain().getEndpoint());
+        mapParameter.put(SystemConstant.ACCESS_KEY_ID, dictionaryConfig.aliYunOssDomain().getAccessKeyId());
+        mapParameter.put(SystemConstant.ACCESS_KEY_SECRET, dictionaryConfig.aliYunOssDomain().getAccessKeySecret());
+        mapParameter.put(SystemConstant.BUCKET, dictionaryConfig.aliYunOssDomain().getBucket());
+        mapParameter.put(SystemConstant.OSS, dictionaryConfig.sysDomain().isOss());
+        mapParameter.put(SystemConstant.ATTACHMENT_TYPE, dictionaryConfig.sysDomain().getAttachmentType());
+        String uploadType = UploadFileEnum.convertToName(type);
+        mapParameter.put(SystemConstant.UPLOAD_TYPE, uploadType);
+        return mapParameter;
+    }
+}

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

@@ -0,0 +1,145 @@
+package com.qmth.themis.task.listener;
+
+import com.alibaba.fastjson.JSONObject;
+import com.google.gson.Gson;
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.entity.TMRocketMessage;
+import com.qmth.themis.business.enums.MqEnum;
+import com.qmth.themis.business.service.TEUserLogService;
+import com.qmth.themis.business.service.TMRocketMessageService;
+import com.qmth.themis.business.templete.TaskImportTemplete;
+import com.qmth.themis.business.templete.impl.ExamStudentTemplete;
+import com.qmth.themis.business.util.JacksonUtil;
+import com.qmth.themis.business.util.OssUtil;
+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.task.constant.TaskConstant;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+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.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.StringJoiner;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @Description: 普通消息监听 task
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/2
+ */
+@Service
+public class RocketTaskConsumer implements MessageListenerConcurrently {
+    private final static Logger log = LoggerFactory.getLogger(RocketTaskConsumer.class);
+
+    @Resource
+    RedisUtil redisUtil;
+
+    @Resource
+    TMRocketMessageService tmRocketMessageService;
+
+    @Resource
+    OssUtil ossUtil;
+
+    @Override
+    @Transactional
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        MqDto mqDto = null;
+        try {
+            long threadId = Thread.currentThread().getId();
+            String threadName = Thread.currentThread().getName();
+            Gson gson = new Gson();
+            for (MessageExt messageExt : msgs) {
+                log.info(":{}-:{} taskConsumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+//                MqDto mqDto = (MqDto) toJavaObject(parseObject(new String(messageExt.getBody(), Constants.CHARSET)), MqDto.class);
+                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+                log.info(":{}-:{} taskConsumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
+                log.info(":{}-:{} taskConsumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
+                if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
+                    Map<String, Object> map = (Map<String, Object>) mqDto.getBody();
+                    JSONObject jsonObject = JSONObject.parseObject(String.valueOf(map.get("remark")));
+                    String type = String.valueOf(jsonObject.get("type"));
+                    String path = String.valueOf(jsonObject.get("path"));
+                    Integer uploadType = Integer.parseInt(String.valueOf(jsonObject.get("uploadType")));
+                    StringJoiner localPath = new StringJoiner("").add(
+                            System.getProperty(SystemConstant.USER_DIR)).add(File.separator).add(path);
+                    File file = null;
+                    if (Objects.nonNull(type) && Objects.equals(type, SystemConstant.LOCAL)) {
+                        file = new File(localPath.toString());
+                    } else {
+                        file = ossUtil.ossDownload(TaskConstant.getAttachmentEnv(uploadType), localPath.toString(), localPath.toString());
+                    }
+                    TaskImportTemplete examStudentTemplete = new ExamStudentTemplete();
+                    examStudentTemplete.importTask(file, map);
+                    mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
+                    mqDto.setBody(JacksonUtil.parseJson(mqDto.getBody()));
+                    TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+                    tmRocketMessageService.saveOrUpdate(tmRocketMessage);
+                    redisUtil.deleteMqTopicList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId());
+                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                } else {
+                    log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
+                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+        }
+        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+    }
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.taskConsumerExamStudentGroup}", topic = "${mq.config.taskTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.taskTopicExamStudentTag}")
+    public class taskConsumerExamStudent 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(RocketTaskConsumer.this::consumeMessage);
+        }
+    }
+
+//    @Service
+//    @RocketMQMessageListener(consumerGroup = "${mq.config.userLogConsumerStudentGroup}", topic = "${mq.config.userLogTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.userLogTopicStudentTag}")
+//    public class sessionConsumerStudentLog 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(RocketTaskConsumer.this::consumeMessage);
+//        }
+//    }
+}

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

@@ -94,6 +94,19 @@ spring.redis.jedis.timeout=180000
 spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
 spring.jackson.time-zone=GMT+8
 
+#\u7CFB\u7EDF\u914D\u7F6E
+sys.config.oss=false
+sys.config.attachmentType=.xlsx,.xls,.doc,.docx,.pdf,.jpg,.jpeg,.png,.html,.zip
+sys.config.serverUpload=/Users/king/git/themis-server/
+spring.resources.static-locations=file:${sys.config.serverUpload},classpath:/META-INF/resources/,classpath:/resources/
+
+#\u963F\u91CC\u4E91OSS\u914D\u7F6E
+aliyun.oss.name=oss-cn-shenzhen.aliyuncs.com
+aliyun.oss.endpoint=http://${aliyun.oss.name}
+aliyun.oss.accessKeyId=LTAI4Fi8jVRYT49QBXU9x5QX
+aliyun.oss.accessKeySecret=97aBLBfkQR5mzCiQa82yWLAH57eUd8
+aliyun.oss.bucket=teachcloud-test
+aliyun.oss.url=http://${aliyun.oss.bucket}.${aliyun.oss.name}
 #============================================================================
 # \u914D\u7F6EJobStore
 #============================================================================
@@ -154,32 +167,15 @@ rocketmq.producer.retry-times-when-send-failed=3
 #ACK
 rocketmq.producer.access-key=AK
 rocketmq.producer.secret-key=SK
+#\u542F\u7528\u6D88\u606F\u8F68\u8FF9\uFF0C\u9ED8\u8BA4\u503Ctrue
 rocketmq.producer.enable-msg-trace=true
+#\u81EA\u5B9A\u4E49\u7684\u6D88\u606F\u8F68\u8FF9\u4E3B\u9898
 rocketmq.producer.customized-trace-topic=my-trace-topic
 
 mq.config.server=themis
-#session_topic\u76D1\u542C
-mq.config.sessionTopic=${mq.config.server}-topic-session
-mq.config.sessionConsumerGroup=${mq.config.server}-group-session
+#task_topic\u76D1\u542C
+mq.config.taskTopic=${mq.config.server}-topic-task
+mq.config.taskConsumerGroup=${mq.config.server}-group-task
 
-mq.config.sessionTopicWebTag=web
-mq.config.sessionConsumerWebGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicWebTag}
-
-mq.config.sessionTopicWxappVideoTag=wxapp_video
-mq.config.sessionConsumerWxappVideoGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicWxappVideoTag}
-
-mq.config.sessionTopicWxappAnswerTag=wxapp_answer
-mq.config.sessionConsumerWxappAnswerGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicWxappAnswerTag}
-
-mq.config.sessionTopicPcTag=pc
-mq.config.sessionConsumerPcGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicPcTag}
-
-#user_login\u76D1\u542C
-mq.config.userLogTopic=${mq.config.server}-topic-userLog
-mq.config.userLogConsumerGroup=${mq.config.server}-group-userLog
-
-mq.config.userLogTopicUserTag=user
-mq.config.userLogConsumerUserGroup=${mq.config.userLogConsumerGroup}-${mq.config.userLogTopicUserTag}
-
-mq.config.userLogTopicStudentTag=student
-mq.config.userLogConsumerStudentGroup=${mq.config.userLogConsumerGroup}-${mq.config.userLogTopicStudentTag}
+mq.config.taskTopicExamStudentTag=examStudent
+mq.config.taskConsumerExamStudentGroup=${mq.config.taskConsumerGroup}-${mq.config.taskTopicExamStudentTag}