mq修改

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

@@ -76,7 +76,7 @@ public class StartRunning implements CommandLineRunner {
         /**
          * websocket mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketUnNormalConsumerOeGroup.getCode(), MqTopicEnum.websocketUnNormalTopic.getCode(), MqTagEnum.oe.name(), MessageModel.CLUSTERING, new WebsocketUnNormalConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketConsumerUnNormalGroup.getCode(), MqTopicEnum.websocketTopic.getCode(), MqTagEnum.unNormal.name(), MessageModel.CLUSTERING, new WebsocketUnNormalConcurrentlyImpl());
         /**
          * websocket mq end
          */

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

@@ -109,7 +109,6 @@ public class WebSocketAdminServer
                         addOnlineCount();
                         //在线数加1
                     }
-                    redisUtil.delete(SystemConstant.WEBSOCKET_UN_NORMAL_LIST, this.sessionId);
                     //发送恢复网络mq消息
                     log.info("用户连接:" + this.sessionId + ",当前在线人数为:" + getOnlineCount());
                     try {
@@ -148,7 +147,6 @@ public class WebSocketAdminServer
             //大于等于超时时间，说明规定时间内都没有通信，非正常退出，因为期间会有心跳更新updateTime
             if ((now.getTime() - this.updateTime) / 1000 >= SystemConstant.WEBSOCKET_MAX_TIME_OUT / 1000) {
                 log.info("超时退出");
-                redisUtil.set(SystemConstant.WEBSOCKET_UN_NORMAL_LIST, this.sessionId, this.sessionId);
                 //发送延时mq消息start
                 MqDtoService mqDtoService = SpringContextHolder.getBean(MqDtoService.class);
                 String level = "2m";
@@ -157,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.websocketUnNormalTopic.getCode(), MqTagEnum.oe.name(), SystemOperationEnum.OE_NET_UN_NORMAL, MqEnum.WEBSOCKET_UN_NORMAL_LOG, this.sessionId, this.tranMap, this.sessionId);
+                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);
                 mqDtoService.assembleSendAsyncDelayMsg(mqDto);
                 //发送延时mq消息end
             }

themis-business/src/main/java/com/qmth/themis/business/cache/ExamRecordCacheUtil.java

@@ -41,5 +41,8 @@ public class ExamRecordCacheUtil {
 	public static Double getObjectiveScore(Long recordId) {
 		return (Double) redisUtil.get(RedisKeyHelper.examRecordCacheKey(recordId), "objectiveScore");
 	}
-	
+
+	public static Date getClientLastSyncTime(Long recordId) {
+		return (Date) redisUtil.get(RedisKeyHelper.examRecordCacheKey(recordId), "clientLastSyncTime");
+	}
 }

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

@@ -125,7 +125,6 @@ public class SystemConstant {
      * websocket
      */
     public static final String WEBSOCKET_OE_ONLINE_COUNT = "websocket:oe:online:count";
-    public static final String WEBSOCKET_UN_NORMAL_LIST = "websocket:oe:unnormal:list";
     public static final String GET = "get";
     public static final long WEBSOCKET_MAX_TIME_OUT = 3 * 60 * 1000;
     //        public static final long WEBSOCKET_MAX_TIME_OUT = 10 * 1000;

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

@@ -50,6 +50,16 @@ public enum MqEnum {
      */
     QUARTZ_LOG(7, "quartz任务"),
 
+    /**
+     * websocket强行交卷
+     */
+    WEBSOCKET_OFFLINE_LOG(8, "websocket强行离线(交卷)"),
+
+    /**
+     * websocket im
+     */
+    WEBSOCKET_IM_LOG(9, "websocket发送消息"),
+
     /**
      * websocket超时退出
      */
@@ -64,62 +74,6 @@ public enum MqEnum {
         this.code = code;
     }
 
-    /**
-     * 状态转换 toId
-     *
-     * @param value
-     * @return
-     */
-    public static int convertToId(String value) {
-        if (Objects.equals(value.trim(), SESSION.name())) {
-            return SESSION.getId();
-        } else if (Objects.equals(value.trim(), EXAM_STUDENT_LOG.name())) {
-            return EXAM_STUDENT_LOG.getId();
-        } else if (Objects.equals(value.trim(), WARMING_LOG.name())) {
-            return WARMING_LOG.getId();
-        } else if (Objects.equals(value.trim(), EXCEPTION_LOG.name())) {
-            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 if (Objects.equals(value.trim(), WEBSOCKET_UN_NORMAL_LOG.name())) {
-            return WEBSOCKET_UN_NORMAL_LOG.getId();
-        } else if (Objects.equals(value.trim(), QUARTZ_LOG.name())) {
-            return QUARTZ_LOG.getId();
-        } else {
-            return MESSAGE_LOG.getId();
-        }
-    }
-
-    /**
-     * 状态转换 toName
-     *
-     * @param value
-     * @return
-     */
-    public static String convertToName(String value) {
-        if (Objects.equals(value.trim(), SESSION.getCode())) {
-            return SESSION.name();
-        } else if (Objects.equals(value.trim(), EXAM_STUDENT_LOG.getCode())) {
-            return EXAM_STUDENT_LOG.name();
-        } else if (Objects.equals(value.trim(), WARMING_LOG.getCode())) {
-            return WARMING_LOG.name();
-        } else if (Objects.equals(value.trim(), EXCEPTION_LOG.getCode())) {
-            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 if (Objects.equals(value.trim(), WEBSOCKET_UN_NORMAL_LOG.getCode())) {
-            return WEBSOCKET_UN_NORMAL_LOG.name();
-        } else if (Objects.equals(value.trim(), QUARTZ_LOG.getCode())) {
-            return QUARTZ_LOG.name();
-        } else {
-            return MESSAGE_LOG.name();
-        }
-    }
-
     public int getId() {
         return id;
     }

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

@@ -74,7 +74,7 @@ public class StartRunning implements CommandLineRunner {
         /**
          * websocket mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketUnNormalConsumerOeGroup.getCode(), MqTopicEnum.websocketUnNormalTopic.getCode(), MqTagEnum.oe.name(), MessageModel.CLUSTERING, new WebsocketUnNormalConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketConsumerUnNormalGroup.getCode(), MqTopicEnum.websocketTopic.getCode(), MqTagEnum.unNormal.name(), MessageModel.CLUSTERING, new WebsocketUnNormalConcurrentlyImpl());
         /**
          * websocket mq end
          */

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

@@ -6,12 +6,15 @@ import com.qmth.themis.business.constant.SpringContextHolder;
 import com.qmth.themis.business.constant.SystemConstant;
 import com.qmth.themis.business.dto.WebsocketDto;
 import com.qmth.themis.business.entity.TBSession;
+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.enums.WebsocketTypeEnum;
+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.business.util.WebsocketUtil;
+import com.qmth.themis.common.contanst.Constants;
 import com.qmth.themis.common.enums.ExceptionResultEnum;
 import com.qmth.themis.common.exception.BusinessException;
 import com.qmth.themis.common.signature.SignatureInfo;
@@ -21,9 +24,22 @@ import com.qmth.themis.mq.dto.MqDto;
 import com.qmth.themis.mq.enums.MqTagEnum;
 import com.qmth.themis.mq.enums.MqTopicEnum;
 import com.qmth.themis.mq.service.MqDtoService;
+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.common.protocol.heartbeat.MessageModel;
+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.Component;
+import org.springframework.stereotype.Service;
 
 import javax.websocket.*;
 import javax.websocket.server.PathParam;
@@ -46,8 +62,7 @@ import java.util.concurrent.ConcurrentHashMap;
 @ServerEndpoint("/oe")
 @Component
 public class WebSocketOeServer
-//        implements MessageListenerConcurrently
-{
+        implements MessageListenerConcurrently {
     private final static Logger log = LoggerFactory.getLogger(WebSocketOeServer.class);
     private volatile static ConcurrentHashMap<String, WebSocketOeServer> webSocketMap = new ConcurrentHashMap<>();
     /**
@@ -109,7 +124,6 @@ public class WebSocketOeServer
                         addOnlineCount();
                         //在线数加1
                     }
-                    redisUtil.delete(SystemConstant.WEBSOCKET_UN_NORMAL_LIST, this.sessionId);
                     //发送恢复网络mq消息
                     log.info("用户连接:" + this.sessionId + ",当前在线人数为:" + getOnlineCount());
                     try {
@@ -149,7 +163,6 @@ public class WebSocketOeServer
             //大于等于超时时间，说明规定时间内都没有通信，非正常退出，因为期间会有心跳更新updateTime
             if ((now.getTime() - this.updateTime) / 1000 >= SystemConstant.WEBSOCKET_MAX_TIME_OUT / 1000) {
                 log.info("超时退出");
-                redisUtil.set(SystemConstant.WEBSOCKET_UN_NORMAL_LIST, this.sessionId, this.sessionId);
                 //发送延时mq消息start
                 MqDtoService mqDtoService = SpringContextHolder.getBean(MqDtoService.class);
                 String level = "2m";
@@ -158,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.websocketUnNormalTopic.getCode(), MqTagEnum.oe.name(), SystemOperationEnum.OE_NET_UN_NORMAL, MqEnum.WEBSOCKET_UN_NORMAL_LOG, String.valueOf(this.recordId), this.tranMap, this.sessionId);
+                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);
                 mqDtoService.assembleSendAsyncDelayMsg(mqDto);
                 //发送延时mq消息end
             }
@@ -275,12 +288,42 @@ public class WebSocketOeServer
         }
     }
 
-//    @Override
-//    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-//        try {
-//            long threadId = Thread.currentThread().getId();
-//            String threadName = Thread.currentThread().getName();
-//            for (MessageExt messageExt : msgs) {
+    @Override
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.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.info(":{}-:{} websocket oe Consumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+                log.info(":{}-:{} websocket oe Consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
+                int reconsumeTime = messageExt.getReconsumeTimes();
+                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
+                    //超过最大重试次数，保存到数据库，后续可以发短信通知系统管理人员
+                    mqDto.setAck(SystemConstant.POSION_ACK_TYPE);
+                    TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+                    tmRocketMessageService.saveOrUpdate(tmRocketMessage);
+                    redisUtil.delete(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId());
+                } else {
+                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
+                        String body = new String(messageExt.getBody(), Constants.CHARSET_NAME);
+                        log.info("body:{}", body);
+                        JSONObject jsonObject = JSONObject.parseObject(body);
+                        Map properties = (Map) jsonObject.get("properties");
+                        String oper = String.valueOf(properties.get("oper"));
+                        if (oper.contains("offLine")) {//下线
+
+                        } else if (oper.contains(MessageModel.CLUSTERING.name())) {//点对点消息
+
+                        } else if (oper.contains(MessageModel.BROADCASTING.name())) {//广播消息
+
+                        }
+                    }
+                }
 //                log.info(":{}-:{} websocketConsumer 重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
 //                MqDto mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
 //                log.info(":{}-:{} websocketConsumer 接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
@@ -301,31 +344,37 @@ public class WebSocketOeServer
 //                        }
 //                    });
 //                }
-//            }
-//        } catch (Exception e) {
-//            e.printStackTrace();
-//            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-//        }
-//        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
-//    }
-//
-//    @Service
-//    @RocketMQMessageListener(consumerGroup = "websocketConsumerImGroup", topic = "websocketImTopic", selectorType = SelectorType.TAG, selectorExpression = "*")
-//    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(WebSocketServer.this::consumeMessage);
-//        }
-//    }
+            }
+        } 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;//成功
+    }
+
+    /**
+     * websocket oe
+     */
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.websocketConsumerOeGroup}", topic = "${mq.config.websocketTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.websocketTopicOeTag}")
+    public class taskConsumerQuartzStudent implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+
+        @Override
+        public void onMessage(Message message) {
+            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
+        }
+
+        @Override
+        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
+            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
+            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
+            defaultMQPushConsumer.registerMessageListener(WebSocketOeServer.this::consumeMessage);
+        }
+    }
 }
 

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

@@ -119,6 +119,11 @@ rocketmq.producer.enable-msg-trace=true
 #\u81EA\u5B9A\u4E49\u7684\u6D88\u606F\u8F68\u8FF9\u4E3B\u9898
 rocketmq.producer.customized-trace-topic=my-trace-topic
 
+#websocket
+mq.config.websocketTopic=themis-topic-websocket
+mq.config.websocketConsumerOeGroup=themis-group-websocket-oe
+mq.config.websocketTopicOeTag=oe
+
 #\u963F\u91CC\u4E91OSS\u914D\u7F6E
 aliyun.oss.name=oss-cn-shenzhen.aliyuncs.com
 aliyun.oss.endpoint=http://${aliyun.oss.name}

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

@@ -72,7 +72,12 @@ public enum MqGroupEnum {
     /**
      * websocket超时退出 考生 group
      */
-    websocketUnNormalConsumerOeGroup("themis-group-websocketUnNormal-oe"),
+    websocketConsumerOeGroup("themis-group-websocket-oe"),
+
+    /**
+     * websocket超时退出 考生 group
+     */
+    websocketConsumerUnNormalGroup("themis-group-websocket-unNormal"),
 
     /**
      * quartz 考场 group

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

@@ -22,5 +22,6 @@ public enum MqTagEnum {
     roomCodeImport,
     examPaperImport,
     oe,
+    unNormal,
     examActivity;
 }

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

@@ -25,9 +25,9 @@ public enum MqTopicEnum {
     taskTopic("themis-topic-task"),
 
     /**
-     * websocketUnNormal topic
+     * websocket topic
      */
-    websocketUnNormalTopic("themis-topic-websocketUnNormal"),
+    websocketTopic("themis-topic-websocket"),
 
     /**
      * quartz topic

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

@@ -2,6 +2,7 @@ package com.qmth.themis.mq.templete.impl;
 
 import com.alibaba.fastjson.JSONObject;
 import com.google.gson.Gson;
+import com.qmth.themis.business.cache.ExamRecordCacheUtil;
 import com.qmth.themis.business.constant.SpringContextHolder;
 import com.qmth.themis.business.constant.SystemConstant;
 import com.qmth.themis.business.entity.TMRocketMessage;
@@ -78,34 +79,39 @@ public class WebsocketUnNormalConcurrentlyImpl implements Concurrently {
                         mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);//表示成功处理
                         Map<String, Object> tranMap = mqDto.getProperties();
                         Long recordId = Long.parseLong(String.valueOf(tranMap.get("recordId")));
-                        String deviceId = String.valueOf(tranMap.get("deviceId"));
-                        String ip = String.valueOf(tranMap.get("ip"));
-                        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);
-                        Integer breakCount = tOeExamRecord.getLeftBreakResumeCount();
-                        if (Objects.isNull(breakCount) || breakCount <= 0) {
-                            //todo 没有断点次数，则强制交卷
-                            tOeExamRecord.setStatus(ExamRecordStatusEnum.finished);
-                            tOeExamRecordService.updateById(tOeExamRecord);
-                        } else {
-                            breakCount--;
-                            //增加断点记录
-                            tOeExamBreakHistory = new TOeExamBreakHistory(recordId, new Date(), BreakReasonEnum.NET_TIME_OUT, BreakReasonEnum.NET_TIME_OUT.name());
-                            tOeExamBreakHistoryService.save(tOeExamBreakHistory);
-                            //更新考试记录状态
-                            tOeExamRecord.setClientCurrentIp(ip);
-                            tOeExamRecord.setClientWebsocketId(deviceId);
-                            tOeExamRecord.setClientWebsocketStatus(WebsocketStatusEnum.UN_ONLINE.ordinal());
-                            tOeExamRecord.setClientLastSyncTime(date);
-                            tOeExamRecord.setLastBreakId(tOeExamBreakHistory.getId());
-                            tOeExamRecord.setLastBreakTime(tOeExamBreakHistory.getBreakTime());
-                            tOeExamRecord.setLeftBreakResumeCount(breakCount);
-                            tOeExamRecord.setStatus(ExamRecordStatusEnum.break_off);
-                            tOeExamRecordService.updateById(tOeExamRecord);
+                        Date clientLastSyncTime = ExamRecordCacheUtil.getClientLastSyncTime(recordId);
+                        if ((System.currentTimeMillis() - clientLastSyncTime.getTime()) / 1000 / 60 >= 2) {//大于等于当前时间，说明未重连或重登录
+                            String deviceId = String.valueOf(tranMap.get("deviceId"));
+                            String ip = String.valueOf(tranMap.get("ip"));
+                            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);
+                            Integer breakCount = tOeExamRecord.getLeftBreakResumeCount();
+                            if (Objects.isNull(breakCount) || breakCount <= 0) {
+                                //todo 没有断点次数，则强制交卷
+                                tOeExamRecord.setStatus(ExamRecordStatusEnum.finished);
+                                tOeExamRecordService.updateById(tOeExamRecord);
+
+                                //加入踢下线mq
+                            } else {
+                                breakCount--;
+                                //增加断点记录
+                                tOeExamBreakHistory = new TOeExamBreakHistory(recordId, new Date(), BreakReasonEnum.NET_TIME_OUT, BreakReasonEnum.NET_TIME_OUT.name());
+                                tOeExamBreakHistoryService.save(tOeExamBreakHistory);
+                                //更新考试记录状态
+                                tOeExamRecord.setClientCurrentIp(ip);
+                                tOeExamRecord.setClientWebsocketId(deviceId);
+                                tOeExamRecord.setClientWebsocketStatus(WebsocketStatusEnum.UN_ONLINE.ordinal());
+                                tOeExamRecord.setClientLastSyncTime(date);
+                                tOeExamRecord.setLastBreakId(tOeExamBreakHistory.getId());
+                                tOeExamRecord.setLastBreakTime(tOeExamBreakHistory.getBreakTime());
+                                tOeExamRecord.setLeftBreakResumeCount(breakCount);
+                                tOeExamRecord.setStatus(ExamRecordStatusEnum.break_off);
+                                tOeExamRecordService.updateById(tOeExamRecord);
+                            }
                         }
                     } else {
                         mqDto.setAck(SystemConstant.INDIVIDUAL_ACK_TYPE);//表示考生已重新登录

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

@@ -14,6 +14,7 @@ import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.BeanUtils;
 import org.springframework.scheduling.quartz.QuartzJobBean;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -46,44 +47,74 @@ public class ExamActivityJob extends QuartzJobBean {
     @Transactional
     protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
         log.info("exam_acvitity_job进来了,context:{}", context);
-        JobDetail jobDetail = context.getJobDetail();
-        String key = String.valueOf(jobDetail.getKey());
-        if (Objects.nonNull(key)) {
-            String[] strings = key.split("\\.");
-            QueryWrapper<TEExamActivity> teExamActivityQueryWrapper = new QueryWrapper<>();
-            teExamActivityQueryWrapper.lambda().eq(TEExamActivity::getCode, strings[1]);
-            TEExamActivity teExamActivity = teExamActivityService.getOne(teExamActivityQueryWrapper);
-            if (Objects.nonNull(teExamActivity)) {
-                log.info("key:{}", key);
+        List<TOeExamRecord> tOeExamRecordUpdateList = null;
+        List<TEExamStudent> teExamStudentUpdateList = null;
+        try {
+            JobDetail jobDetail = context.getJobDetail();
+            String key = String.valueOf(jobDetail.getKey());
+            if (Objects.nonNull(key)) {
+                String[] strings = key.split("\\.");
+                QueryWrapper<TEExamActivity> teExamActivityQueryWrapper = new QueryWrapper<>();
+                teExamActivityQueryWrapper.lambda().eq(TEExamActivity::getCode, strings[1]);
+                TEExamActivity teExamActivity = teExamActivityService.getOne(teExamActivityQueryWrapper);
+                if (Objects.nonNull(teExamActivity)) {
+                    log.info("key:{}", key);
 
-                //获取该考试批次下所有未交卷的考生的考试记录
-                QueryWrapper<TOeExamRecord> tOeExamRecordQueryWrapper = new QueryWrapper<>();
-                tOeExamRecordQueryWrapper.lambda().eq(TOeExamRecord::getExamActivityId, teExamActivity.getId()).eq(TOeExamRecord::getStatus, ExamRecordStatusEnum.finished.ordinal());
-                List<TOeExamRecord> tOeExamRecordList = tOeExamRecordService.list(tOeExamRecordQueryWrapper);
-                List<Long> examStudentIdList = new ArrayList<>();
-                tOeExamRecordList.forEach(s -> {
-                    s.setStatus(ExamRecordStatusEnum.finished);
-                    s.setFinishTime(new Date());
-                    s.setFinishType(FinishTypeEnum.INTERRUPT);
-                    examStudentIdList.add(s.getExamStudentId());
-                });
-                tOeExamRecordService.updateBatchById(tOeExamRecordList);
-
-                if (examStudentIdList.size() > 0) {
-                    //获取该考试批次下所有考生，考试次数减1
-                    QueryWrapper<TEExamStudent> teExamStudentQueryWrapper = new QueryWrapper<>();
-                    teExamStudentQueryWrapper.lambda().in(TEExamStudent::getId, examStudentIdList);
-                    List<TEExamStudent> teExamStudentList = teExamStudentService.list(teExamStudentQueryWrapper);
-                    teExamStudentList.forEach(s -> {
-                        int count = s.getLeftExamCount();
-                        count--;
-                        s.setLeftExamCount(count < 0 ? 0 : count);
+                    //获取该考试批次下所有未交卷的考生的考试记录
+                    QueryWrapper<TOeExamRecord> tOeExamRecordQueryWrapper = new QueryWrapper<>();
+                    tOeExamRecordQueryWrapper.lambda().eq(TOeExamRecord::getExamActivityId, teExamActivity.getId()).ne(TOeExamRecord::getStatus, ExamRecordStatusEnum.finished.ordinal());
+                    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);
+                        finalExamStudentIdList.add(s.getExamStudentId());
+                        finalTOeExamRecordUpdateList.add(t);
                     });
-                    teExamStudentService.updateBatchById(teExamStudentList);
+                    tOeExamRecordService.updateBatchById(tOeExamRecordList);
+
+                    if (examStudentIdList.size() > 0) {
+                        //获取该考试批次下所有考生，考试次数减1
+                        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);
+                    }
+                    int i = 1 / 0;
+                    //todo 未完待续,需要加入交卷逻辑
+                } else {
+                    log.info("考试场次:{}已删除", key);
                 }
-                //todo 未完待续,需要加入交卷逻辑
-            } else {
-                log.info("考试场次:{}已删除", key);
+            }
+        } 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);
             }
         }
     }

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

@@ -91,7 +91,7 @@ public class StartRunning implements CommandLineRunner {
         /**
          * websocket mq start
          */
-        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketUnNormalConsumerOeGroup.getCode(), MqTopicEnum.websocketUnNormalTopic.getCode(), MqTagEnum.oe.name(), MessageModel.CLUSTERING, new WebsocketUnNormalConcurrentlyImpl());
+        rocketMessageConsumer.setRocketMQConsumer(nameServer, MqGroupEnum.websocketConsumerUnNormalGroup.getCode(), MqTopicEnum.websocketTopic.getCode(), MqTagEnum.unNormal.name(), MessageModel.CLUSTERING, new WebsocketUnNormalConcurrentlyImpl());
         /**
          * websocket mq end
          */