新加清理缓存执行器

examcloud-core-oe-task-service/src/main/java/cn/com/qmth/examcloud/core/oe/task/service/ExamBossService.java

@@ -0,0 +1,32 @@
+package cn.com.qmth.examcloud.core.oe.task.service;
+
+import cn.com.qmth.examcloud.support.examing.ExamBoss;
+
+/**
+ * @Description 考试基础信息会话
+ * @Author lideyin
+ * @Date 2019/12/20 15:41
+ * @Version 1.0
+ */
+public interface ExamBossService {
+
+    /**
+     *
+     * @param examId
+     * @param eb
+     */
+    public void saveExamBoss(Long examId, ExamBoss eb);
+
+    /**
+     * 获取
+     * @param examId
+     * @return
+     */
+    public ExamBoss getExamBoss(Long examId);
+
+    /**
+     * 删除
+     * @param examId
+     */
+    public void deleteExamBoss(Long examId);
+}

examcloud-core-oe-task-service/src/main/java/cn/com/qmth/examcloud/core/oe/task/service/impl/ExamBossServiceImpl.java

@@ -0,0 +1,39 @@
+package cn.com.qmth.examcloud.core.oe.task.service.impl;
+
+import cn.com.qmth.examcloud.core.oe.task.service.ExamBossService;
+import cn.com.qmth.examcloud.support.examing.ExamBoss;
+import cn.com.qmth.examcloud.support.redis.RedisKeyHelper;
+import cn.com.qmth.examcloud.web.redis.RedisClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * @Description 考试基础信息会话
+ * @Author lideyin
+ * @Date 2019/12/20 15:40
+ * @Version 1.0
+ */
+@Service("examBossService")
+public class ExamBossServiceImpl implements ExamBossService {
+
+	@Autowired
+	private RedisClient redisClient;
+
+	@Override
+	public void saveExamBoss(Long examId, ExamBoss eb) {
+		String key = RedisKeyHelper.getBuilder().examBossKey(examId);
+		redisClient.set(key, eb, -1);
+	}
+
+	@Override
+	public ExamBoss getExamBoss(Long examId) {
+		String key = RedisKeyHelper.getBuilder().examBossKey(examId);
+		return redisClient.get(key, ExamBoss.class);
+	}
+
+	@Override
+	public void deleteExamBoss(Long examId) {
+		String key = RedisKeyHelper.getBuilder().examBossKey(examId);
+		redisClient.delete(key);
+	}
+}

examcloud-core-oe-task-service/src/main/java/cn/com/qmth/examcloud/core/oe/task/service/pipeline/ClearExamDataCacheExecutor.java

@@ -0,0 +1,70 @@
+package cn.com.qmth.examcloud.core.oe.task.service.pipeline;
+
+import cn.com.qmth.examcloud.commons.helpers.KeyValuePair;
+import cn.com.qmth.examcloud.commons.helpers.pipeline.NodeExecuter;
+import cn.com.qmth.examcloud.commons.helpers.pipeline.TaskContext;
+import cn.com.qmth.examcloud.core.oe.admin.api.SyncExamDataCloudService;
+import cn.com.qmth.examcloud.core.oe.student.api.ExamRecordDataCloudService;
+import cn.com.qmth.examcloud.core.oe.task.dao.ExamCaptureRepo;
+import cn.com.qmth.examcloud.core.oe.task.dao.ExamSyncCaptureRepo;
+import cn.com.qmth.examcloud.core.oe.task.service.ExamBossService;
+import cn.com.qmth.examcloud.core.oe.task.service.ExamRecordDataService;
+import cn.com.qmth.examcloud.support.Constants;
+import cn.com.qmth.examcloud.support.enums.SyncStatus;
+import cn.com.qmth.examcloud.support.examing.ExamBoss;
+import cn.com.qmth.examcloud.support.examing.ExamRecordData;
+import cn.com.qmth.examcloud.web.helpers.SequenceLockHelper;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * @Description 清理考试记录缓存
+ * @Author lideyin
+ * @Date 2019/12/24 18:07
+ * @Version 1.0
+ */
+@Component
+public class ClearExamDataCacheExecutor implements NodeExecuter<Long, ExamRecordData, Long, ExamRecordData> {
+
+    @Autowired
+    private ExamBossService examBossService;
+
+    @Override
+    public List<KeyValuePair<Long, ExamRecordData>> execute(Long key, ExamRecordData examRecordData, TaskContext context) throws Exception {
+        String sequenceLockKey = Constants.EXAM_CONTROL_LOCK_PREFIX + examRecordData.getStudentId();
+
+        try {
+            //添加考试控制全局锁
+            SequenceLockHelper.getLock(sequenceLockKey);
+
+            //只有已同步成功的数据，才执行清理操作
+            if (SyncStatus.SYNCED == examRecordData.getSyncStatus()) {
+                clearExamCache(examRecordData.getExamId());
+            }
+
+            return null;
+        } finally {
+            SequenceLockHelper.releaseLock(sequenceLockKey);
+        }
+    }
+
+    /**
+     * 考试id
+     *
+     * @param examId
+     */
+    private void clearExamCache(Long examId) {
+        ExamBoss examBoss = examBossService.getExamBoss(examId);
+
+        if (null == examBoss) {
+            return;
+        }
+
+        //如果开考次数==考试完结次数，则删除考试基础信息缓存
+        if (examBoss.getStartCount() == examBoss.getEndCount()) {
+            examBossService.deleteExamBoss(examId);
+        }
+    }
+}

examcloud-core-oe-task-service/src/main/java/cn/com/qmth/examcloud/core/oe/task/service/pipeline/SyncExamDataExecutor.java

@@ -1,5 +1,6 @@
 package cn.com.qmth.examcloud.core.oe.task.service.pipeline;
 
+import cn.com.qmth.examcloud.commons.exception.StatusException;
 import cn.com.qmth.examcloud.commons.helpers.KeyValuePair;
 import cn.com.qmth.examcloud.commons.helpers.pipeline.NodeExecuter;
 import cn.com.qmth.examcloud.commons.helpers.pipeline.TaskContext;
@@ -20,9 +21,12 @@ import cn.com.qmth.examcloud.core.oe.task.dao.ExamCaptureRepo;
 import cn.com.qmth.examcloud.core.oe.task.dao.ExamSyncCaptureRepo;
 import cn.com.qmth.examcloud.core.oe.task.dao.entity.ExamCaptureEntity;
 import cn.com.qmth.examcloud.core.oe.task.dao.entity.ExamSyncCaptureEntity;
+import cn.com.qmth.examcloud.core.oe.task.service.ExamBossService;
 import cn.com.qmth.examcloud.core.oe.task.service.ExamRecordDataService;
 import cn.com.qmth.examcloud.support.Constants;
 import cn.com.qmth.examcloud.support.enums.ExamRecordStatus;
+import cn.com.qmth.examcloud.support.enums.SyncStatus;
+import cn.com.qmth.examcloud.support.examing.ExamBoss;
 import cn.com.qmth.examcloud.support.examing.ExamRecordData;
 import cn.com.qmth.examcloud.web.helpers.SequenceLockHelper;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -50,6 +54,8 @@ public class SyncExamDataExecutor implements NodeExecuter<Long, ExamRecordData,
     private ExamSyncCaptureRepo examSyncCaptureRepo;
     @Autowired
     private ExamRecordDataCloudService examRecordDataCloudService;
+    @Autowired
+    private ExamBossService examBossService;
 
     @Override
     public List<KeyValuePair<Long, ExamRecordData>> execute(Long key, ExamRecordData examRecordData, TaskContext context) throws Exception {
@@ -59,6 +65,11 @@ public class SyncExamDataExecutor implements NodeExecuter<Long, ExamRecordData,
             //添加考试控制全局锁
             SequenceLockHelper.getLock(sequenceLockKey);
 
+            //如果已同步，直接返回
+            if (SyncStatus.SYNCED == examRecordData.getSyncStatus()) {
+                return null;
+            }
+
             //处理上一节点中，指定时间内仍未处理完成的数据
             Long examRecordDataId = examRecordData.getId();
             if (examRecordData.getExamRecordStatus() == ExamRecordStatus.EXAM_HAND_IN ||
@@ -66,6 +77,9 @@ public class SyncExamDataExecutor implements NodeExecuter<Long, ExamRecordData,
                 examRecordDataService.processAfterHandInExam(examRecordDataId);
             }
 
+            List<KeyValuePair<Long, ExamRecordData>> resultList = new ArrayList<>();
+            KeyValuePair<Long, ExamRecordData> keyValuePair = new KeyValuePair<>(key, examRecordData);
+
             //同步数据
             SyncExamDataReq syncReq = new SyncExamDataReq();
             syncReq.setExamRecordData(copyExamRecordDataFrom(examRecordData));
@@ -77,7 +91,20 @@ public class SyncExamDataExecutor implements NodeExecuter<Long, ExamRecordData,
             syncReq.setFaceBiopsy(getFaceBiopsy(examRecordDataId));
             syncExamDataCloudService.syncExamData(syncReq);
 
-            return null;
+            //考试完结次数加1
+            ExamBoss examBoss = examBossService.getExamBoss(examRecordData.getExamId());
+            if (null != examBoss) {
+                examBoss.setEndCount(examBoss.getEndCount() + 1);
+            }
+
+            //设置并保存考试记录的同步状态
+            examRecordData.setSyncStatus(SyncStatus.SYNCED);
+            setAndSaveExamRecordDataSyncStatus(examRecordDataId);
+
+            keyValuePair.setValue(examRecordData);
+            resultList.add(keyValuePair);
+
+            return resultList;
         } finally {
             SequenceLockHelper.releaseLock(sequenceLockKey);
         }
@@ -310,4 +337,19 @@ public class SyncExamDataExecutor implements NodeExecuter<Long, ExamRecordData,
         return data;
     }
 
+    /**
+     * 设置并保存考试记录的同步状态
+     *
+     * @param examRecordDataId
+     */
+    private void setAndSaveExamRecordDataSyncStatus(Long examRecordDataId) {
+        ExamRecordData examRecordData = examRecordDataService.getExamRecordDataCache(examRecordDataId);
+
+        if (SyncStatus.SYNCED == examRecordData.getSyncStatus()) {
+            return;
+        }
+
+        examRecordData.setSyncStatus(SyncStatus.SYNCED);
+        examRecordDataService.saveExamRecordDataCache(examRecordDataId, examRecordData);
+    }
 }

examcloud-core-oe-task-starter/src/main/java/cn/com/qmth/examcloud/core/oe/task/starter/config/StreamTaskExecutor.java

@@ -3,10 +3,7 @@ package cn.com.qmth.examcloud.core.oe.task.starter.config;
 import cn.com.qmth.examcloud.commons.helpers.pipeline.Node;
 import cn.com.qmth.examcloud.commons.helpers.pipeline.SimpleNode;
 import cn.com.qmth.examcloud.commons.helpers.pipeline.TaskContext;
-import cn.com.qmth.examcloud.core.oe.task.service.pipeline.AfterHandInExamExecutor;
-import cn.com.qmth.examcloud.core.oe.task.service.pipeline.DataGainExamExecutor;
-import cn.com.qmth.examcloud.core.oe.task.service.pipeline.HandInExamExecutor;
-import cn.com.qmth.examcloud.core.oe.task.service.pipeline.SyncExamDataExecutor;
+import cn.com.qmth.examcloud.core.oe.task.service.pipeline.*;
 import cn.com.qmth.examcloud.support.cache.CacheHelper;
 import cn.com.qmth.examcloud.support.cache.bean.SysPropertyCacheBean;
 import cn.com.qmth.examcloud.support.examing.ExamRecordData;
@@ -29,16 +26,19 @@ public class StreamTaskExecutor implements ApplicationRunner {
     @Autowired
     private DataGainExamExecutor dataGainExamExecutor;
     @Autowired
-    HandInExamExecutor handInExamExecutor;
+    private HandInExamExecutor handInExamExecutor;
     @Autowired
-    AfterHandInExamExecutor afterHandInExamExecutor;
+    private AfterHandInExamExecutor afterHandInExamExecutor;
     @Autowired
-    SyncExamDataExecutor syncExamDataExecutor;
+    private SyncExamDataExecutor syncExamDataExecutor;
+    @Autowired
+    private ClearExamDataCacheExecutor clearExamDataCacheExecutor;
 
     private static Integer DEFAULT_GAIN_EXAM_DATA_EXECUTOR_SLEEP_SECONDS = 0;
     private static Integer DEFAULT_HAND_IN_EXAM_EXECUTOR_SLEEP_SECONDS = 5 * 60;
     private static Integer DEFAULT_AFTER_HAND_IN_EXAM_EXECUTOR_SLEEP_SECONDS = 5;
     private static Integer DEFAULT_SYNC_EXAM_DATA_EXECUTOR_SLEEP_SECONDS = 5;
+    private static Integer DEFAULT_CLEAR_EXAM_DATA_CACHE_EXECUTOR_SLEEP_SECONDS = 5;
 
     @Override
     public void run(ApplicationArguments args) throws Exception {
@@ -60,28 +60,34 @@ public class StreamTaskExecutor implements ApplicationRunner {
         Node<Long, ExamRecordData, Long, ExamRecordData> node4 = new SimpleNode<>(
                 "syncExamData", syncExamDataExecutor, context);
 
+        Node<Long, ExamRecordData, Long, ExamRecordData> node5 = new SimpleNode<>(
+                "clearExamDataCache", clearExamDataCacheExecutor, context);
+
         node1.setFirst(true);
         node1.setLowerNode(node2);
         node1.setSleep(PropertyHolder.getInt("oeTask.executor.gainExamData.sleep",
                 DEFAULT_GAIN_EXAM_DATA_EXECUTOR_SLEEP_SECONDS));//单位秒
 
         node2.setLowerNode(node3);
-        SysPropertyCacheBean node2SleepProperty = CacheHelper.getSysProperty("oeTask.executor.handInExam.sleep");
-        node2.setSleep(PropertyHolder.getInt("oeTask.executor.gainExamData.sleep",
+        node2.setSleep(PropertyHolder.getInt("oeTask.executor.handInExam.sleep",
                 DEFAULT_HAND_IN_EXAM_EXECUTOR_SLEEP_SECONDS));//单位秒
 
         node3.setLowerNode(node4);
-        SysPropertyCacheBean node3SleepProperty = CacheHelper.getSysProperty("oeTask.executor.afterHandInExam.sleep");
-        node3.setSleep(PropertyHolder.getInt("oeTask.executor.gainExamData.sleep",
+        node3.setSleep(PropertyHolder.getInt("oeTask.executor.afterHandInExam.sleep",
                 DEFAULT_AFTER_HAND_IN_EXAM_EXECUTOR_SLEEP_SECONDS));//单位秒
 
-        node4.setSleep(PropertyHolder.getInt("oeTask.executor.gainExamData.sleep",
+        node4.setLowerNode(node5);
+        node4.setSleep(PropertyHolder.getInt("oeTask.executor.syncExamData.sleep",
                 DEFAULT_SYNC_EXAM_DATA_EXECUTOR_SLEEP_SECONDS));//单位秒
 
+        node5.setSleep(PropertyHolder.getInt("oeTask.executor.clearExamDataCache.sleep",
+                DEFAULT_CLEAR_EXAM_DATA_CACHE_EXECUTOR_SLEEP_SECONDS));//单位秒
+
         node1.start();
         node2.start();
         node3.start();
         node4.start();
+        node5.start();
     }