mq修改

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

@@ -156,7 +156,7 @@ public class TBExamInvigilateUserController {
             transMap.put("orgId", tbUser.getOrgId());
             transMap.put("remark", tbAttachment.getRemark());
             //mq发送消息start
-            MqDto mqDto = new MqDto(MqTopicEnum.taskTopic.getCode(), MqTagEnum.roomCodeImport.name(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
+            MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.roomCodeImport.name(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
             mqDtoService.assembleSendOneWayMsg(mqDto);
             //mq发送消息end
         } catch (Exception e) {
@@ -193,7 +193,7 @@ public class TBExamInvigilateUserController {
                 transMap.put("createId", tbUser.getId());
                 transMap.put("orgId", tbUser.getOrgId());
                 //mq发送消息start
-                MqDto mqDto = new MqDto(MqTopicEnum.taskTopic.getCode(), MqTagEnum.roomCodeExport.name(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
+                MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.roomCodeExport.name(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
                 mqDtoService.assembleSendOneWayMsg(mqDto);
                 //mq发送消息end
             } else {
@@ -201,6 +201,11 @@ public class TBExamInvigilateUserController {
             }
         } catch (Exception e) {
             e.printStackTrace();
+            if (e instanceof BusinessException) {
+                throw new BusinessException(e.getMessage());
+            } else {
+                throw new RuntimeException(e);
+            }
         }
         Map map = new HashMap();
         map.put(SystemConstant.TASK_ID, tbTaskHistory.getId());

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

@@ -194,8 +194,8 @@ public class TBUserController {
         TBSession tbSession = new TBSession(sessionId, String.valueOf(user.getId()), authDto.getRoleCodes().toString(), source, platform.name(), deviceId, ServletUtil.getRequest().getLocalAddr(), token, expire);
         redisUtil.setUserSession(sessionId, tbSession, redisExpire);
         //mq发送消息start
-        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.sessionTopic.getCode(), platform.name(), tbSession, MqEnum.SESSION.name(), tbSession.getId(), user.getLoginName());
-        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.userLogTopic.getCode(), authDto.getRoleCodes().toString().contains(RoleEnum.STUDENT.name()) ? MqTagEnum.student.name() : MqTagEnum.user.name(), SystemOperationEnum.LOGIN, MqEnum.USER_LOG.name(), user.getId(), user.getLoginName());
+        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.themisTopic.getCode(), platform.name(), tbSession, MqEnum.SESSION.name(), tbSession.getId(), user.getLoginName());
+        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.themisTopic.getCode(), authDto.getRoleCodes().toString().contains(RoleEnum.STUDENT.name()) ? MqTagEnum.student.name() : MqTagEnum.user.name(), SystemOperationEnum.LOGIN, MqEnum.USER_LOG.name(), user.getId(), user.getLoginName());
         //mq发送消息end
         //测试
         String test = SignatureInfo.build(SignatureType.TOKEN, sessionId, token);
@@ -541,7 +541,7 @@ public class TBUserController {
             cacheService.removeAccountCache(tbUser.getId());
         }
         //mq发送消息start
-        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.userLogTopic.getCode(), authDto.getRoleCodes().toString().contains(RoleEnum.STUDENT.name()) ? MqTagEnum.student.name() : MqTagEnum.user.name(), SystemOperationEnum.LOGOUT, MqEnum.USER_LOG.name(), tbUser.getId(), tbUser.getLoginName());
+        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.themisTopic.getCode(), authDto.getRoleCodes().toString().contains(RoleEnum.STUDENT.name()) ? MqTagEnum.student.name() : MqTagEnum.user.name(), SystemOperationEnum.LOGOUT, MqEnum.USER_LOG.name(), tbUser.getId(), tbUser.getLoginName());
         //mq发送消息end
         return ResultUtil.ok(JacksonUtil.parseJson(SystemConstant.SUCCESS));
     }

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

@@ -78,7 +78,7 @@ public class TEExamActivityController {
             Map<String, Object> prop = new HashMap<>();
             prop.put("oper", "insert");
             prop.put("exam", teExam);
-            MqDto mqDto = new MqDto(MqTopicEnum.quartzTopic.getCode(), MqTagEnum.examActivity.name(), JacksonUtil.parseJson(teExamActivityList), MqEnum.QUARTZ_LOG, String.valueOf(teExam.getId()), prop, tbUser.getName());
+            MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.examActivity.name(), JacksonUtil.parseJson(teExamActivityList), MqEnum.QUARTZ_LOG, String.valueOf(teExam.getId()), prop, tbUser.getName());
             mqDtoService.assembleSendOneWayMsg(mqDto);
             //新增quartz任务,发送mq消息end
         } catch (Exception e) {

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

@@ -89,7 +89,7 @@ public class TEExamController {
                 //删除quartz任务,发送mq消息start
                 Map<String, Object> prop = new HashMap<>();
                 prop.put("oper", "delete");
-                MqDto mqDto = new MqDto(MqTopicEnum.quartzTopic.getCode(), MqTagEnum.examActivity.name(), JacksonUtil.parseJson(teExamActivityList), MqEnum.QUARTZ_LOG, String.valueOf(teExam.getId()), prop, tbUser.getName());
+                MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.examActivity.name(), JacksonUtil.parseJson(teExamActivityList), MqEnum.QUARTZ_LOG, String.valueOf(teExam.getId()), prop, tbUser.getName());
                 mqDtoService.assembleSendOneWayMsg(mqDto);
                 //删除quartz任务,发送mq消息end
 
@@ -106,7 +106,7 @@ public class TEExamController {
                     Map<String, Object> prop = new HashMap<>();
                     prop.put("oper", "insert");
                     prop.put("exam", teExam);
-                    MqDto mqDto = new MqDto(MqTopicEnum.quartzTopic.getCode(), MqTagEnum.examActivity.name(), JacksonUtil.parseJson(Arrays.asList(teExamActivity)), MqEnum.QUARTZ_LOG, String.valueOf(teExam.getId()), prop, tbUser.getName());
+                    MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.examActivity.name(), JacksonUtil.parseJson(Arrays.asList(teExamActivity)), MqEnum.QUARTZ_LOG, String.valueOf(teExam.getId()), prop, tbUser.getName());
                     mqDtoService.assembleSendOneWayMsg(mqDto);
                     //新增quartz任务,发送mq消息end
                 } else {//修改需删除之前生成的全部quartz任务
@@ -124,7 +124,7 @@ public class TEExamController {
                     Map<String, Object> prop = new HashMap<>();
                     prop.put("oper", "insert");
                     prop.put("exam", teExam);
-                    MqDto mqDto = new MqDto(MqTopicEnum.quartzTopic.getCode(), MqTagEnum.examActivity.name(), JacksonUtil.parseJson(teExamActivityList), MqEnum.QUARTZ_LOG, String.valueOf(teExam.getId()), prop, tbUser.getName());
+                    MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.examActivity.name(), JacksonUtil.parseJson(teExamActivityList), MqEnum.QUARTZ_LOG, String.valueOf(teExam.getId()), prop, tbUser.getName());
                     mqDtoService.assembleSendOneWayMsg(mqDto);
                     //新增quartz任务,发送mq消息end
                 }

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

@@ -178,7 +178,7 @@ public class TEExamPaperController {
             transMap.put("processPaper", processPaper);
             transMap.put("processAnswer", processAnswer);
             //mq发送消息start
-            MqDto mqDto = new MqDto(MqTopicEnum.taskTopic.getCode(), MqTagEnum.examPaperImport.name(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
+            MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.examPaperImport.name(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
             mqDtoService.assembleSendOneWayMsg(mqDto);
             //mq发送消息end
         } catch (Exception e) {

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

@@ -196,7 +196,7 @@ public class TEExamStudentController {
             }
             transMap.put("remark", tbAttachment.getRemark());
             //mq发送消息start
-            MqDto mqDto = new MqDto(MqTopicEnum.taskTopic.getCode(), MqTagEnum.examStudentImport.name(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
+            MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.examStudentImport.name(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
             mqDtoService.assembleSendOneWayMsg(mqDto);
             //mq发送消息end
         } catch (Exception e) {

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

@@ -1,15 +1,12 @@
 package com.qmth.themis.backend.start;
 
-import com.qmth.themis.business.config.SystemConfig;
+import com.qmth.themis.business.constant.SpringContextHolder;
 import com.qmth.themis.business.constant.SystemConstant;
 import com.qmth.themis.mq.enums.MqGroupEnum;
 import com.qmth.themis.mq.enums.MqTagEnum;
 import com.qmth.themis.mq.enums.MqTopicEnum;
 import com.qmth.themis.mq.listener.RocketMessageConsumer;
-import com.qmth.themis.mq.templete.impl.SessionConcurrentlyImpl;
-import com.qmth.themis.mq.templete.impl.TaskConcurrentlyImpl;
-import com.qmth.themis.mq.templete.impl.UserLogConcurrentlyImpl;
-import com.qmth.themis.mq.templete.impl.WebsocketUnNormalConcurrentlyImpl;
+import com.qmth.themis.mq.templete.impl.*;
 import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -18,7 +15,6 @@ import org.springframework.boot.CommandLineRunner;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
-import java.io.File;
 
 /**
  * @Description: 服务启动时初始化运行，哪个微服务模块需要则拿此模版去用
@@ -43,12 +39,12 @@ public class StartRunning implements CommandLineRunner {
         /**
          * session mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWebGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.web.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWinGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.win.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerMacGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.mac.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWxappGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.wxapp.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerIosGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.ios.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerAndroidGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.android.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWebGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.web.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWinGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.win.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerMacGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.mac.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWxappGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.wxapp.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerIosGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.ios.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerAndroidGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.android.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
         /**
          * session mq end
          */
@@ -56,8 +52,8 @@ public class StartRunning implements CommandLineRunner {
         /**
          * userLog mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerUserGroup.getCode(), MqTopicEnum.userLogTopic.getCode(), MqTagEnum.user.name(), MessageModel.CLUSTERING, new UserLogConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerStudentGroup.getCode(), MqTopicEnum.userLogTopic.getCode(), MqTagEnum.student.name(), MessageModel.CLUSTERING, new UserLogConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerUserGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.user.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(UserLogConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerStudentGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.student.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(UserLogConcurrentlyImpl.class));
         /**
          * userLog mq end
          */
@@ -65,10 +61,10 @@ public class StartRunning implements CommandLineRunner {
         /**
          * task mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamStudentImportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.examStudentImport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeImportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.roomCodeImport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeExportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.roomCodeExport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamPaperImportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.examPaperImport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamStudentImportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.examStudentImport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeImportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.roomCodeImport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeExportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.roomCodeExport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamPaperImportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.examPaperImport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
         /**
          * task mq end
          */
@@ -76,10 +72,12 @@ public class StartRunning implements CommandLineRunner {
         /**
          * websocket mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketConsumerUnNormalGroup.getCode(), MqTopicEnum.websocketTopic.getCode(), MqTagEnum.unNormal.name(), MessageModel.CLUSTERING, new WebsocketUnNormalConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketConsumerUnNormalGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.unNormal.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(WebsocketUnNormalConcurrentlyImpl.class));
         /**
          * websocket mq end
          */
+        //计算客观分
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.normalGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.calculateObjectiveScore.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(CalculateObjectiveScoreConcurrentlyImpl.class));
         SystemConstant.initTempFiles();
         log.info("服务器启动时执行 end");
     }

themis-backend/src/main/java/com/qmth/themis/backend/websocket/WebSocketAdminServer.java

@@ -155,7 +155,7 @@ public class WebSocketAdminServer
                 dt = dt.plusMinutes(Long.parseLong(level.replace("m", "")));
                 tranMap.put("timeOut", time);
                 tranMap.put("mqExecTime", dt.toInstant(ZoneOffset.of("+8")).toEpochMilli());
-                MqDto mqDto = new MqDto(MqTopicEnum.websocketTopic.getCode(), MqTagEnum.unNormal.name(), SystemOperationEnum.OE_NET_UN_NORMAL, MqEnum.WEBSOCKET_UN_NORMAL_LOG, this.sessionId, this.tranMap, this.sessionId);
+                MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.unNormal.name(), SystemOperationEnum.OE_NET_UN_NORMAL, MqEnum.WEBSOCKET_UN_NORMAL_LOG, this.sessionId, this.tranMap, this.sessionId);
                 mqDtoService.assembleSendAsyncDelayMsg(mqDto);
                 //发送延时mq消息end
             }

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

@@ -233,7 +233,7 @@ public class TEExamController {
         transMap.put("subNumber", subNumber);
         transMap.put("subIndex", subIndex);
         //mq发送消息start
-        MqDto mqDto = new MqDto(MqTopicEnum.normal.getCode(), MqTagEnum.calculateObjectiveScore.name(), transMap, MqEnum.EXAM, null, null);
+        MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.calculateObjectiveScore.name(), transMap, MqEnum.EXAM, null, null);
         mqDtoService.assembleSendOneWayMsg(mqDto);
 	}
 	

themis-exam/src/main/java/com/qmth/themis/exam/api/TEStudentController.java

@@ -159,8 +159,8 @@ public class TEStudentController {
         TBSession tbSession = new TBSession(sessionId, String.valueOf(teStudent.getId()), authDto.getRoleCodes().toString(), source, platform.name(), deviceId, ServletUtil.getRequest().getLocalAddr(), token, expire);
         redisUtil.setUserSession(sessionId, tbSession, redisExpire);
         //mq发送消息start
-        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.sessionTopic.getCode(), platform.name(), tbSession, MqEnum.SESSION.name(), tbSession.getId(), teStudent.getIdentity());
-        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.userLogTopic.getCode(), authDto.getRoleCodes().toString().contains(RoleEnum.STUDENT.name()) ? MqTagEnum.student.name() : MqTagEnum.user.name(), SystemOperationEnum.LOGIN, MqEnum.EXAM_STUDENT_LOG.name(), teStudent.getId(), teStudent.getIdentity());
+        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.themisTopic.getCode(), platform.name(), tbSession, MqEnum.SESSION.name(), tbSession.getId(), teStudent.getIdentity());
+        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.themisTopic.getCode(), authDto.getRoleCodes().toString().contains(RoleEnum.STUDENT.name()) ? MqTagEnum.student.name() : MqTagEnum.user.name(), SystemOperationEnum.LOGIN, MqEnum.EXAM_STUDENT_LOG.name(), teStudent.getId(), teStudent.getIdentity());
         //mq发送消息end
         //测试
         String test = SignatureInfo.build(SignatureType.TOKEN, sessionId, token);
@@ -210,7 +210,7 @@ public class TEStudentController {
             cacheService.removeStudentCache(teStudent.getId());
         }
         //mq发送消息start
-        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.userLogTopic.getCode(), authDto.getRoleCodes().toString().contains(RoleEnum.STUDENT.name()) ? MqTagEnum.student.name() : MqTagEnum.user.name(), SystemOperationEnum.LOGOUT, MqEnum.EXAM_STUDENT_LOG.name(), teStudent.getId(), teStudent.getIdentity());
+        mqDtoService.assembleSendOneWayMsg(MqTopicEnum.themisTopic.getCode(), authDto.getRoleCodes().toString().contains(RoleEnum.STUDENT.name()) ? MqTagEnum.student.name() : MqTagEnum.user.name(), SystemOperationEnum.LOGOUT, MqEnum.EXAM_STUDENT_LOG.name(), teStudent.getId(), teStudent.getIdentity());
         //mq发送消息end
         return ResultUtil.ok(JacksonUtil.parseJson(SystemConstant.SUCCESS));
     }

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

@@ -2,6 +2,7 @@ package com.qmth.themis.exam.start;
 
 import javax.annotation.Resource;
 
+import com.qmth.themis.business.constant.SpringContextHolder;
 import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -43,12 +44,12 @@ public class StartRunning implements CommandLineRunner {
         /**
          * session mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWebGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.web.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWinGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.win.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerMacGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.mac.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWxappGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.wxapp.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerIosGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.ios.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerAndroidGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.android.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWebGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.web.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWinGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.win.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerMacGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.mac.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWxappGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.wxapp.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerIosGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.ios.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerAndroidGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.android.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
         /**
          * session mq end
          */
@@ -56,8 +57,8 @@ public class StartRunning implements CommandLineRunner {
         /**
          * userLog mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerUserGroup.getCode(), MqTopicEnum.userLogTopic.getCode(), MqTagEnum.user.name(), MessageModel.CLUSTERING, new UserLogConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerStudentGroup.getCode(), MqTopicEnum.userLogTopic.getCode(), MqTagEnum.student.name(), MessageModel.CLUSTERING, new UserLogConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerUserGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.user.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(UserLogConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerStudentGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.student.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(UserLogConcurrentlyImpl.class));
         /**
          * userLog mq end
          */
@@ -65,10 +66,10 @@ public class StartRunning implements CommandLineRunner {
         /**
          * task mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamStudentImportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.examStudentImport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeImportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.roomCodeImport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeExportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.roomCodeExport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamPaperImportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.examPaperImport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamStudentImportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.examStudentImport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeImportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.roomCodeImport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeExportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.roomCodeExport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamPaperImportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.examPaperImport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
         /**
          * task mq end
          */
@@ -76,13 +77,12 @@ public class StartRunning implements CommandLineRunner {
         /**
          * websocket mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketConsumerUnNormalGroup.getCode(), MqTopicEnum.websocketTopic.getCode(), MqTagEnum.unNormal.name(), MessageModel.CLUSTERING, new WebsocketUnNormalConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketConsumerUnNormalGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.unNormal.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(WebsocketUnNormalConcurrentlyImpl.class));
         /**
          * websocket mq end
          */
-        
         //计算客观分
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.normalGroup.getCode(), MqTopicEnum.normal.getCode(), MqTagEnum.calculateObjectiveScore.name(), MessageModel.CLUSTERING, new CalculateObjectiveScoreConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.normalGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.calculateObjectiveScore.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(CalculateObjectiveScoreConcurrentlyImpl.class));
         SystemConstant.initTempFiles();
         log.info("服务器启动时执行 end");
     }

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

@@ -171,7 +171,7 @@ public class WebSocketOeServer
                 dt = dt.plusMinutes(Long.parseLong(level.replace("m", "")));
                 tranMap.put("timeOut", time);
                 tranMap.put("mqExecTime", dt.toInstant(ZoneOffset.of("+8")).toEpochMilli());
-                MqDto mqDto = new MqDto(MqTopicEnum.websocketTopic.getCode(), MqTagEnum.unNormal.name(), SystemOperationEnum.OE_NET_UN_NORMAL, MqEnum.WEBSOCKET_UN_NORMAL_LOG, String.valueOf(this.recordId), this.tranMap, this.sessionId);
+                MqDto mqDto = new MqDto(MqTopicEnum.themisTopic.getCode(), MqTagEnum.unNormal.name(), SystemOperationEnum.OE_NET_UN_NORMAL, MqEnum.WEBSOCKET_UN_NORMAL_LOG, String.valueOf(this.recordId), this.tranMap, this.sessionId);
                 mqDtoService.assembleSendAsyncDelayMsg(mqDto);
                 //发送延时mq消息end
             }

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

@@ -120,8 +120,8 @@ rocketmq.producer.enable-msg-trace=true
 rocketmq.producer.customized-trace-topic=my-trace-topic
 
 #websocket
-mq.config.websocketTopic=themis-topic-websocket
-mq.config.websocketConsumerOeGroup=themis-group-websocket-oe
+mq.config.websocketTopic=themis-topic-exam
+mq.config.websocketConsumerOeGroup=themis-group-exam-websocketOe
 mq.config.websocketTopicOeTag=oe
 
 #\u963F\u91CC\u4E91OSS\u914D\u7F6E

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

@@ -12,82 +12,82 @@ public enum MqGroupEnum {
     /**
      * web group
      */
-    sessionConsumerWebGroup("themis-group-session-web"),
+    sessionConsumerWebGroup("themis-group-exam-sessionWeb"),
 
     /**
      * win group
      */
-    sessionConsumerWinGroup("themis-group-session-win"),
+    sessionConsumerWinGroup("themis-group-exam-sessionWin"),
 
     /**
      * mac group
      */
-    sessionConsumerMacGroup("themis-group-session-mac"),
+    sessionConsumerMacGroup("themis-group-exam-sessionMac"),
 
     /**
      * wxapp group
      */
-    sessionConsumerWxappGroup("themis-group-session-wxapp"),
+    sessionConsumerWxappGroup("themis-group-exam-sessionWxapp"),
 
     /**
      * ios group
      */
-    sessionConsumerIosGroup("themis-group-session-ios"),
+    sessionConsumerIosGroup("themis-group-exam-sessionIos"),
 
     /**
      * android group
      */
-    sessionConsumerAndroidGroup("themis-group-session-android"),
+    sessionConsumerAndroidGroup("themis-group-exam-sessionAndroid"),
 
     /**
      * 用户轨迹 user group
      */
-    userLogConsumerUserGroup("themis-group-userLog-user"),
+    userLogConsumerUserGroup("themis-group-exam-userLogUser"),
 
     /**
      * 用户轨迹 student group
      */
-    userLogConsumerStudentGroup("themis-group-userLog-student"),
+    userLogConsumerStudentGroup("themis-group-exam-userLogStudent"),
 
     /**
      * 异步任务 考生导入 group
      */
-    taskConsumerExamStudentImportGroup("themis-group-task-examStudentImport"),
+    taskConsumerExamStudentImportGroup("themis-group-exam-taskExamStudentImport"),
 
     /**
      * 异步任务 考场导出 group
      */
-    taskConsumerRoomCodeExportGroup("themis-group-task-roomCodeExport"),
+    taskConsumerRoomCodeExportGroup("themis-group-exam-taskRoomCodeExport"),
 
     /**
      * 异步任务 考场导入 group
      */
-    taskConsumerRoomCodeImportGroup("themis-group-task-roomCodeImport"),
+    taskConsumerRoomCodeImportGroup("themis-group-exam-taskRoomCodeImport"),
 
     /**
      * 异步任务 试卷导入 group
      */
-    taskConsumerExamPaperImportGroup("themis-group-task-examPaperImport"),
+    taskConsumerExamPaperImportGroup("themis-group-exam-taskExamPaperImport"),
 
     /**
      * websocket超时退出 考生 group
      */
-    websocketConsumerOeGroup("themis-group-websocket-oe"),
+    websocketConsumerOeGroup("themis-group-exam-websocketOe"),
 
     /**
      * websocket超时退出 考生 group
      */
-    websocketConsumerUnNormalGroup("themis-group-websocket-unNormal"),
+    websocketConsumerUnNormalGroup("themis-group-exam-websocketUnNormal"),
 
     /**
      * quartz 考场 group
      */
-    quartzConsumerExamActivityGroup("themis-group-quartz-examActivity"),
+    quartzConsumerExamActivityGroup("themis-group-exam-quartzExamActivity"),
 	
     /**
      * 通用分组
      */
-    normalGroup("themis-group-normal");
+    normalGroup("themis-group-exam-normal");
 
     private MqGroupEnum(String code) {
         this.code = code;

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

@@ -9,37 +9,9 @@ package com.qmth.themis.mq.enums;
  */
 public enum MqTopicEnum {
 
-    /**
-     * session topic
-     */
-    sessionTopic("themis-topic-session"),
+    themisTopic("themis-topic-exam");
 
-    /**
-     * userLog topic
-     */
-    userLogTopic("themis-topic-userLog"),
-
-    /**
-     * task topic
-     */
-    taskTopic("themis-topic-task"),
-
-    /**
-     * websocket topic
-     */
-    websocketTopic("themis-topic-websocket"),
-
-    /**
-     * quartz topic
-     */
-    quartzTopic("themis-topic-quartz"),
-	
-	/**
-	 * 通用消息
-	 */
-	normal("themis-topic-normal");
-
-    private MqTopicEnum(String code){
+    private MqTopicEnum(String code) {
         this.code = code;
     }
 

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

@@ -1,58 +0,0 @@
-package com.qmth.themis.mq.listener;
-
-import com.qmth.themis.business.constant.SystemConstant;
-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.message.Message;
-import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
-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 java.util.List;
-
-/**
- * @Description: 事务消息监听
- * @Param:
- * @return:
- * @Author: wangliang
- * @Date: 2020/6/28
- */
-@Service
-public class RocketConsumerTranListener {
-    private final static Logger log = LoggerFactory.getLogger(RocketConsumerTranListener.class);
-
-    @Service
-    @RocketMQMessageListener(consumerGroup = "themis-tran-group", topic = "themis-tran-topic")
-    public class consumerTranListener implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
-
-        @Override
-        public void onMessage(Message message) {
-            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
-            log.info("实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用");
-        }
-
-        @Override
-        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
-            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
-            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
-            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
-                @Override
-                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
-                    for (MessageExt messageExt : msgs) {
-                        log.info("重试次数:{}", messageExt.getReconsumeTimes());
-                        log.info("接受到的消息:{}", new String(messageExt.getBody()));
-                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                    }
-                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
-//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-                }
-            });
-        }
-    }
-}

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

@@ -1,17 +1,5 @@
 package com.qmth.themis.mq.templete.impl;
 
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
-import org.apache.rocketmq.common.message.MessageExt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
-
 import com.google.gson.Gson;
 import com.qmth.themis.business.constant.SpringContextHolder;
 import com.qmth.themis.business.constant.SystemConstant;
@@ -21,73 +9,89 @@ import com.qmth.themis.business.service.TOeExamRecordService;
 import com.qmth.themis.business.util.JacksonUtil;
 import com.qmth.themis.business.util.RedisUtil;
 import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.common.exception.BusinessException;
 import com.qmth.themis.mq.dto.MqDto;
 import com.qmth.themis.mq.templete.Concurrently;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 
 /**
  * 计算客观分
- * 
+ *
  * @Description:
  * @Author: xiatian
  * @Date: 2020-07-30
  */
 @Service
 public class CalculateObjectiveScoreConcurrentlyImpl implements Concurrently {
-	private final static Logger log = LoggerFactory.getLogger(CalculateObjectiveScoreConcurrentlyImpl.class);
+    private final static Logger log = LoggerFactory.getLogger(CalculateObjectiveScoreConcurrentlyImpl.class);
 
-	private RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
-	private TOeExamRecordService examRecordService = SpringContextHolder.getBean(TOeExamRecordService.class);
-	private TMRocketMessageService tmRocketMessageService = SpringContextHolder.getBean(TMRocketMessageService.class);
-	@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.debug(":{}-:{} CalculateObjectiveScore 重试次数:{}", threadId, threadName,
-						messageExt.getReconsumeTimes());
-				mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-				log.debug(":{}-:{} CalculateObjectiveScore 接收到的消息:{}", threadId, threadName,
-						JacksonUtil.parseJson(mqDto));
-				int reconsumeTime = messageExt.getReconsumeTimes();
-				if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
-					 //超过最大重试次数，保存到数据库，后续可以发短信通知系统管理人员
+    @Override
+    @Transactional
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
+                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
+        TOeExamRecordService examRecordService = SpringContextHolder.getBean(TOeExamRecordService.class);
+        TMRocketMessageService tmRocketMessageService = SpringContextHolder.getBean(TMRocketMessageService.class);
+        MqDto mqDto = null;
+        try {
+            long threadId = Thread.currentThread().getId();
+            String threadName = Thread.currentThread().getName();
+            Gson gson = new Gson();
+            for (MessageExt messageExt : msgs) {
+                log.debug(":{}-:{} CalculateObjectiveScore 重试次数:{}", threadId, threadName,
+                        messageExt.getReconsumeTimes());
+                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+                log.debug(":{}-:{} CalculateObjectiveScore 接收到的消息:{}", threadId, threadName,
+                        JacksonUtil.parseJson(mqDto));
+                int reconsumeTime = messageExt.getReconsumeTimes();
+                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
+                    //超过最大重试次数，保存到数据库，后续可以发短信通知系统管理人员
                     mqDto.setAck(SystemConstant.POSION_ACK_TYPE);
                     TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
                     tmRocketMessageService.saveOrUpdate(tmRocketMessage);
                     redisUtil.delete(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId());
-				} else {
-					if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
-							&& Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
-							&& redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
-									SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
-						log.debug(":{}-:{} 更新db", threadId, threadName);
-						Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
-						examRecordService.calculateObjectiveScore(param);
-						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.delete(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId());
-						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-					} else {
-						log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-						return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-					}
-				}
-			}
-		} catch (Exception e) {
-			return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
-		} finally {
-			if (Objects.nonNull(mqDto)) {
-				redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
-			}
-		}
-		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
-	}
+                } else {
+                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
+                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
+                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
+                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
+                        log.debug(":{}-:{} 更新db", threadId, threadName);
+                        Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
+                        examRecordService.calculateObjectiveScore(param);
+                        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.delete(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId());
+                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                    } else {
+                        log.debug(":{}-:{} 消息ack未确认，重发", threadId, threadName);
+                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
+                    }
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            if (e instanceof BusinessException) {
+                throw new BusinessException(e.getMessage());
+            } else {
+                throw new RuntimeException(e);
+            }
+        } finally {
+            if (Objects.nonNull(mqDto)) {
+                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
+            }
+        }
+        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
+    }
 }

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

@@ -10,6 +10,7 @@ import com.qmth.themis.business.service.TMRocketMessageService;
 import com.qmth.themis.business.util.JacksonUtil;
 import com.qmth.themis.business.util.RedisUtil;
 import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.common.exception.BusinessException;
 import com.qmth.themis.mq.dto.MqDto;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
@@ -77,7 +78,11 @@ public class SessionConcurrentlyImpl implements Concurrently {
             }
         } catch (Exception e) {
             e.printStackTrace();
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+            if (e instanceof BusinessException) {
+                throw new BusinessException(e.getMessage());
+            } else {
+                throw new RuntimeException(e);
+            }
         } finally {
             if (Objects.nonNull(mqDto)) {
                 redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());

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

@@ -19,6 +19,7 @@ import com.qmth.themis.business.threadPool.MyThreadPool;
 import com.qmth.themis.business.util.JacksonUtil;
 import com.qmth.themis.business.util.RedisUtil;
 import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.common.exception.BusinessException;
 import com.qmth.themis.mq.dto.MqDto;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
@@ -113,7 +114,11 @@ public class TaskConcurrentlyImpl implements Concurrently {
             }
         } catch (Exception e) {
             e.printStackTrace();
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+            if (e instanceof BusinessException) {
+                throw new BusinessException(e.getMessage());
+            } else {
+                throw new RuntimeException(e);
+            }
         } finally {
             if (Objects.nonNull(mqDto)) {
                 redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());

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

@@ -12,6 +12,7 @@ import com.qmth.themis.business.service.TMRocketMessageService;
 import com.qmth.themis.business.util.JacksonUtil;
 import com.qmth.themis.business.util.RedisUtil;
 import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.common.exception.BusinessException;
 import com.qmth.themis.mq.dto.MqDto;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
@@ -82,7 +83,11 @@ public class UserLogConcurrentlyImpl implements Concurrently {
             }
         } catch (Exception e) {
             e.printStackTrace();
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+            if (e instanceof BusinessException) {
+                throw new BusinessException(e.getMessage());
+            } else {
+                throw new RuntimeException(e);
+            }
         } finally {
             if (Objects.nonNull(mqDto)) {
                 redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());

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

@@ -16,6 +16,7 @@ import com.qmth.themis.business.service.TOeExamRecordService;
 import com.qmth.themis.business.util.JacksonUtil;
 import com.qmth.themis.business.util.RedisUtil;
 import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.common.exception.BusinessException;
 import com.qmth.themis.mq.dto.MqDto;
 import com.qmth.themis.mq.templete.Concurrently;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
@@ -23,7 +24,6 @@ import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -52,9 +52,6 @@ public class WebsocketUnNormalConcurrentlyImpl implements Concurrently {
         TOeExamRecordService tOeExamRecordService = SpringContextHolder.getBean(TOeExamRecordService.class);
         TOeExamBreakHistoryService tOeExamBreakHistoryService = SpringContextHolder.getBean(TOeExamBreakHistoryService.class);
         MqDto mqDto = null;
-        TOeExamRecord tOeExamRecord = null;
-        TOeExamBreakHistory tOeExamBreakHistory = null;
-        TOeExamRecord tOeExamRecordUpdate = null;
         try {
             long threadId = Thread.currentThread().getId();
             String threadName = Thread.currentThread().getName();
@@ -86,9 +83,7 @@ public class WebsocketUnNormalConcurrentlyImpl implements Concurrently {
                             Long updateTime = Long.parseLong(String.valueOf(tranMap.get("updateTime")));
                             Date date = new Date();
                             date.setTime(updateTime);
-                            tOeExamRecord = tOeExamRecordService.getById(recordId);
-                            tOeExamRecordUpdate = new TOeExamRecord();
-                            BeanUtils.copyProperties(tOeExamRecord, tOeExamRecordUpdate);
+                            TOeExamRecord tOeExamRecord = tOeExamRecordService.getById(recordId);
                             Integer breakCount = tOeExamRecord.getLeftBreakResumeCount();
                             if (Objects.isNull(breakCount) || breakCount <= 0) {
                                 //todo 没有断点次数，则强制交卷
@@ -99,7 +94,7 @@ public class WebsocketUnNormalConcurrentlyImpl implements Concurrently {
                             } else {
                                 breakCount--;
                                 //增加断点记录
-                                tOeExamBreakHistory = new TOeExamBreakHistory(recordId, new Date(), BreakReasonEnum.NET_TIME_OUT, BreakReasonEnum.NET_TIME_OUT.name());
+                                TOeExamBreakHistory tOeExamBreakHistory = new TOeExamBreakHistory(recordId, new Date(), BreakReasonEnum.NET_TIME_OUT, BreakReasonEnum.NET_TIME_OUT.name());
                                 tOeExamBreakHistoryService.save(tOeExamBreakHistory);
                                 //更新考试记录状态
                                 tOeExamRecord.setClientCurrentIp(ip);
@@ -126,13 +121,11 @@ public class WebsocketUnNormalConcurrentlyImpl implements Concurrently {
             }
         } catch (Exception e) {
             e.printStackTrace();
-            if (Objects.nonNull(tOeExamRecordUpdate)) {
-                tOeExamRecordService.updateById(tOeExamRecordUpdate);
+            if (e instanceof BusinessException) {
+                throw new BusinessException(e.getMessage());
+            } else {
+                throw new RuntimeException(e);
             }
-            if (Objects.nonNull(tOeExamBreakHistory)) {
-                tOeExamBreakHistoryService.removeById(tOeExamBreakHistory.getId());
-            }
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
         } finally {
             if (Objects.nonNull(mqDto)) {
                 redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());

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

@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.google.gson.Gson;
+import com.qmth.themis.business.constant.SpringContextHolder;
 import com.qmth.themis.business.constant.SystemConstant;
 import com.qmth.themis.business.entity.TEExam;
 import com.qmth.themis.business.entity.TEExamActivity;
@@ -12,6 +13,7 @@ import com.qmth.themis.business.service.TMRocketMessageService;
 import com.qmth.themis.business.util.JacksonUtil;
 import com.qmth.themis.business.util.RedisUtil;
 import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.common.exception.BusinessException;
 import com.qmth.themis.mq.dto.MqDto;
 import com.qmth.themis.task.config.DictionaryConfig;
 import com.qmth.themis.task.quartz.ExamActivityJob;
@@ -119,7 +121,7 @@ public class RocketQuartzConsumer implements MessageListenerOrderly {
                                         Map mapJob = new HashMap();
                                         mapJob.put("name", activityCode);
                                         quartzService.deleteJob(activityCode, dictionaryConfig.quartzConfigDomain().getExamActivityJobGroupName());
-                                        quartzService.addJob(ExamActivityJob.class, activityCode, dictionaryConfig.quartzConfigDomain().getExamActivityJobGroupName(), cron, mapJob);
+                                        quartzService.addJob(SpringContextHolder.getBean(ExamActivityJob.class).getClass(), activityCode, dictionaryConfig.quartzConfigDomain().getExamActivityJobGroupName(), cron, mapJob);
                                     }
                                 }
                             }
@@ -136,7 +138,11 @@ public class RocketQuartzConsumer implements MessageListenerOrderly {
             }
         } catch (Exception e) {
             e.printStackTrace();
-            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+            if (e instanceof BusinessException) {
+                throw new BusinessException(e.getMessage());
+            } else {
+                throw new RuntimeException(e);
+            }
         } finally {
             if (Objects.nonNull(mqDto)) {
                 redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());

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

@@ -9,6 +9,7 @@ import com.qmth.themis.business.enums.FinishTypeEnum;
 import com.qmth.themis.business.service.TEExamActivityService;
 import com.qmth.themis.business.service.TEExamStudentService;
 import com.qmth.themis.business.service.TOeExamRecordService;
+import com.qmth.themis.common.exception.BusinessException;
 import org.quartz.JobDetail;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
@@ -47,8 +48,6 @@ public class ExamActivityJob extends QuartzJobBean {
     @Transactional
     protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
         log.info("exam_acvitity_job进来了,context:{}", context);
-        List<TOeExamRecord> tOeExamRecordUpdateList = null;
-        List<TEExamStudent> teExamStudentUpdateList = null;
         try {
             JobDetail jobDetail = context.getJobDetail();
             String key = String.valueOf(jobDetail.getKey());
@@ -66,19 +65,14 @@ public class ExamActivityJob extends QuartzJobBean {
                     List<TOeExamRecord> tOeExamRecordList = tOeExamRecordService.list(tOeExamRecordQueryWrapper);
                     List<Long> examStudentIdList = null;
                     if (Objects.nonNull(tOeExamRecordList) && tOeExamRecordList.size() > 0) {
-                        tOeExamRecordUpdateList = new ArrayList<>();
                         examStudentIdList = new ArrayList<>();
                     }
                     List<Long> finalExamStudentIdList = examStudentIdList;
-                    List<TOeExamRecord> finalTOeExamRecordUpdateList = tOeExamRecordUpdateList;
                     tOeExamRecordList.forEach(s -> {
-                        TOeExamRecord t = new TOeExamRecord();
-                        BeanUtils.copyProperties(s, t);
                         s.setStatus(ExamRecordStatusEnum.finished);
                         s.setFinishTime(new Date());
-                        s.setFinishType(FinishTypeEnum.INTERRUPT);
+                        s.setFinishType(FinishTypeEnum.interrupt);
                         finalExamStudentIdList.add(s.getExamStudentId());
-                        finalTOeExamRecordUpdateList.add(t);
                     });
                     tOeExamRecordService.updateBatchById(tOeExamRecordList);
 
@@ -87,17 +81,10 @@ public class ExamActivityJob extends QuartzJobBean {
                         QueryWrapper<TEExamStudent> teExamStudentQueryWrapper = new QueryWrapper<>();
                         teExamStudentQueryWrapper.lambda().in(TEExamStudent::getId, examStudentIdList);
                         List<TEExamStudent> teExamStudentList = teExamStudentService.list(teExamStudentQueryWrapper);
-                        if (Objects.nonNull(teExamStudentList) && teExamStudentList.size() > 0) {
-                            teExamStudentUpdateList = new ArrayList<>();
-                        }
-                        List<TEExamStudent> finalTeExamStudentUpdateList = teExamStudentUpdateList;
                         teExamStudentList.forEach(s -> {
-                            TOeExamRecord t = new TOeExamRecord();
-                            BeanUtils.copyProperties(s, t);
                             int count = Objects.isNull(s.getLeftExamCount()) ? 0 : s.getLeftExamCount();
                             count--;
                             s.setLeftExamCount(count < 0 ? 0 : count);
-                            finalTeExamStudentUpdateList.add(s);
                             //加入踢下线mq
                         });
                         teExamStudentService.updateBatchById(teExamStudentList);
@@ -110,11 +97,10 @@ public class ExamActivityJob extends QuartzJobBean {
             }
         } catch (Exception e) {
             e.printStackTrace();
-            if (Objects.nonNull(tOeExamRecordUpdateList) && tOeExamRecordUpdateList.size() > 0) {
-                tOeExamRecordService.updateBatchById(tOeExamRecordUpdateList);
-            }
-            if (Objects.nonNull(teExamStudentUpdateList) && teExamStudentUpdateList.size() > 0) {
-                teExamStudentService.updateBatchById(teExamStudentUpdateList);
+            if (e instanceof BusinessException) {
+                throw new BusinessException(e.getMessage());
+            } else {
+                throw new RuntimeException(e);
             }
         }
     }

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

@@ -1,13 +1,11 @@
 package com.qmth.themis.task.start;
 
+import com.qmth.themis.business.constant.SpringContextHolder;
 import com.qmth.themis.mq.enums.MqGroupEnum;
 import com.qmth.themis.mq.enums.MqTagEnum;
 import com.qmth.themis.mq.enums.MqTopicEnum;
 import com.qmth.themis.mq.listener.RocketMessageConsumer;
-import com.qmth.themis.mq.templete.impl.SessionConcurrentlyImpl;
-import com.qmth.themis.mq.templete.impl.TaskConcurrentlyImpl;
-import com.qmth.themis.mq.templete.impl.UserLogConcurrentlyImpl;
-import com.qmth.themis.mq.templete.impl.WebsocketUnNormalConcurrentlyImpl;
+import com.qmth.themis.mq.templete.impl.*;
 import com.qmth.themis.task.config.DictionaryConfig;
 import com.qmth.themis.task.quartz.MqJob;
 import com.qmth.themis.task.service.QuartzService;
@@ -58,12 +56,12 @@ public class StartRunning implements CommandLineRunner {
         /**
          * session mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWebGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.web.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWinGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.win.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerMacGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.mac.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWxappGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.wxapp.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerIosGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.ios.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerAndroidGroup.getCode(), MqTopicEnum.sessionTopic.getCode(), MqTagEnum.android.name(), MessageModel.CLUSTERING, new SessionConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWebGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.web.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWinGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.win.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerMacGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.mac.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerWxappGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.wxapp.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerIosGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.ios.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.sessionConsumerAndroidGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.android.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(SessionConcurrentlyImpl.class));
         /**
          * session mq end
          */
@@ -71,8 +69,8 @@ public class StartRunning implements CommandLineRunner {
         /**
          * userLog mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerUserGroup.getCode(), MqTopicEnum.userLogTopic.getCode(), MqTagEnum.user.name(), MessageModel.CLUSTERING, new UserLogConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerStudentGroup.getCode(), MqTopicEnum.userLogTopic.getCode(), MqTagEnum.student.name(), MessageModel.CLUSTERING, new UserLogConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerUserGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.user.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(UserLogConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.userLogConsumerStudentGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.student.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(UserLogConcurrentlyImpl.class));
         /**
          * userLog mq end
          */
@@ -80,10 +78,10 @@ public class StartRunning implements CommandLineRunner {
         /**
          * task mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamStudentImportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.examStudentImport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeImportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.roomCodeImport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeExportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.roomCodeExport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamPaperImportGroup.getCode(), MqTopicEnum.taskTopic.getCode(), MqTagEnum.examPaperImport.name(), MessageModel.CLUSTERING, new TaskConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamStudentImportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.examStudentImport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeImportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.roomCodeImport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerRoomCodeExportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.roomCodeExport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.taskConsumerExamPaperImportGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.examPaperImport.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(TaskConcurrentlyImpl.class));
         /**
          * task mq end
          */
@@ -91,10 +89,12 @@ public class StartRunning implements CommandLineRunner {
         /**
          * websocket mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketConsumerUnNormalGroup.getCode(), MqTopicEnum.websocketTopic.getCode(), MqTagEnum.unNormal.name(), MessageModel.CLUSTERING, new WebsocketUnNormalConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketConsumerUnNormalGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.unNormal.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(WebsocketUnNormalConcurrentlyImpl.class));
         /**
          * websocket mq end
          */
+        //计算客观分
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.normalGroup.getCode(), MqTopicEnum.themisTopic.getCode(), MqTagEnum.calculateObjectiveScore.name(), MessageModel.CLUSTERING, SpringContextHolder.getBean(CalculateObjectiveScoreConcurrentlyImpl.class));
         log.info("服务器启动时执行 end");
     }
 }

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

@@ -173,6 +173,6 @@ rocketmq.producer.enable-msg-trace=true
 rocketmq.producer.customized-trace-topic=my-trace-topic
 
 #quartz-\u8003\u8BD5\u573A\u6B21task
-mq.config.quartzTopic=themis-topic-quartz
-mq.config.quartzConsumerExamActivityGroup=themis-group-quartz-examActivity
+mq.config.quartzTopic=themis-topic-exam
+mq.config.quartzConsumerExamActivityGroup=themis-group-exam-quartzExamActivity
 mq.config.quartzTopicExamActivityTag=examActivity