考务导入接口

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

@@ -1,6 +1,5 @@
 package com.qmth.themis.backend.api;
 
-import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -200,7 +199,7 @@ public class TEExamStudentController {
             }
             transMap.put("remark", tbAttachment.getRemark());
             //mq发送消息start
-            MqDto mqDto = new MqDto(dictionaryConfig.mqConfigDomain().getTaskTopic(), dictionaryConfig.mqConfigDomain().getTaskTopicExamStudentTag(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
+            MqDto mqDto = new MqDto(dictionaryConfig.mqConfigDomain().getTaskTopic(), dictionaryConfig.mqConfigDomain().getTaskTopicExamStudentImportTag(), transMap, MqEnum.TASK_LOG, String.valueOf(tbTaskHistory.getId()), tbUser.getName());
             mqDtoService.assembleSendOneWayMsg(mqDto);
             //mq发送消息end
         } catch (Exception e) {

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

@@ -185,8 +185,11 @@ mq.config.userLogConsumerStudentGroup=${mq.config.userLogConsumerGroup}-${mq.con
 mq.config.taskTopic=${mq.config.server}-topic-task
 mq.config.taskConsumerGroup=${mq.config.server}-group-task
 
-mq.config.taskTopicExamStudentTag=examStudent
-mq.config.taskConsumerExamStudentGroup=${mq.config.taskConsumerGroup}-${mq.config.taskTopicExamStudentTag}
+mq.config.taskTopicExamStudentImportTag=examStudentImport
+mq.config.taskConsumerExamStudentImportGroup=${mq.config.taskConsumerGroup}-${mq.config.taskTopicExamStudentImportTag}
+
+mq.config.taskTopicRoomCodeImportTag=roomCodeImport
+mq.config.taskConsumerRoomCodeImportGroup=${mq.config.taskConsumerGroup}-${mq.config.taskTopicRoomCodeImportTag}
 
 #dlq\u6B7B\u4FE1\u961F\u5217
 #mq.config.sessionConsumerGroupDlq=${mq.config.sessionConsumerGroup}-dlq

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

@@ -59,6 +59,7 @@ public class SystemConstant {
     public static final String UPLOAD_TYPE = "uploadType";
     public static final String DEFAULT_PASSWORD = "yXVUkR45PFz0UfpbDB8/ew==";
     public static final String USER_DIR = "user.dir";
+    public static final int MAX_IMPORT_SIZE = 500;
     /**
      * session过期时间
      */

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

@@ -14,7 +14,8 @@ public class MqConfigDomain {
     private String sessionConsumerGroup;
     //    private String sessionConsumerGroupDlq;
 //    private String sessionTopicDlq;
-    /**taskTopicExamStudentTag
+    /**
+     * taskTopicExamStudentTag
      * session topic
      */
     private String sessionTopicWebTag;
@@ -41,39 +42,57 @@ public class MqConfigDomain {
      */
     private String taskTopic;
     private String taskConsumerGroup;
-    private String taskTopicExamStudentTag;
-    private String taskConsumerExamStudentGroup;
+    private String taskTopicExamStudentImportTag;
+    private String taskConsumerExamStudentImportGroup;
+    private String taskTopicRoomCodeImportTag;
+    private String taskConsumerRoomCodeImportGroup;
 
-    public String getTaskTopic() {
-        return taskTopic;
+    public String getTaskTopicExamStudentImportTag() {
+        return taskTopicExamStudentImportTag;
     }
 
-    public void setTaskTopic(String taskTopic) {
-        this.taskTopic = taskTopic;
+    public void setTaskTopicExamStudentImportTag(String taskTopicExamStudentImportTag) {
+        this.taskTopicExamStudentImportTag = taskTopicExamStudentImportTag;
     }
 
-    public String getTaskConsumerGroup() {
-        return taskConsumerGroup;
+    public String getTaskConsumerExamStudentImportGroup() {
+        return taskConsumerExamStudentImportGroup;
     }
 
-    public void setTaskConsumerGroup(String taskConsumerGroup) {
-        this.taskConsumerGroup = taskConsumerGroup;
+    public void setTaskConsumerExamStudentImportGroup(String taskConsumerExamStudentImportGroup) {
+        this.taskConsumerExamStudentImportGroup = taskConsumerExamStudentImportGroup;
+    }
+
+    public String getTaskTopicRoomCodeImportTag() {
+        return taskTopicRoomCodeImportTag;
     }
 
-    public String getTaskTopicExamStudentTag() {
-        return taskTopicExamStudentTag;
+    public void setTaskTopicRoomCodeImportTag(String taskTopicRoomCodeImportTag) {
+        this.taskTopicRoomCodeImportTag = taskTopicRoomCodeImportTag;
     }
 
-    public void setTaskTopicExamStudentTag(String taskTopicExamStudentTag) {
-        this.taskTopicExamStudentTag = taskTopicExamStudentTag;
+    public String getTaskConsumerRoomCodeImportGroup() {
+        return taskConsumerRoomCodeImportGroup;
     }
 
-    public String getTaskConsumerExamStudentGroup() {
-        return taskConsumerExamStudentGroup;
+    public void setTaskConsumerRoomCodeImportGroup(String taskConsumerRoomCodeImportGroup) {
+        this.taskConsumerRoomCodeImportGroup = taskConsumerRoomCodeImportGroup;
     }
 
-    public void setTaskConsumerExamStudentGroup(String taskConsumerExamStudentGroup) {
-        this.taskConsumerExamStudentGroup = taskConsumerExamStudentGroup;
+    public String getTaskTopic() {
+        return taskTopic;
+    }
+
+    public void setTaskTopic(String taskTopic) {
+        this.taskTopic = taskTopic;
+    }
+
+    public String getTaskConsumerGroup() {
+        return taskConsumerGroup;
+    }
+
+    public void setTaskConsumerGroup(String taskConsumerGroup) {
+        this.taskConsumerGroup = taskConsumerGroup;
     }
 
     public String getSessionTopicWxappAnswerTag() {

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

@@ -1,11 +0,0 @@
-package com.qmth.themis.business.templete;
-
-/** 
-* @Description: 导出任务回调
-* @Param:  
-* @return:  
-* @Author: wangliang
-* @Date: 2020/7/15 
-*/ 
-public interface TaskExportCallback {
-}

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

@@ -1,11 +0,0 @@
-package com.qmth.themis.business.templete;
-
-/** 
-* @Description: 导入任务回调
-* @Param:  
-* @return:  
-* @Author: wangliang
-* @Date: 2020/7/15 
-*/ 
-public interface TaskImportCallback {
-}

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

@@ -0,0 +1,177 @@
+package com.qmth.themis.business.templete;
+
+import com.alibaba.fastjson.JSONObject;
+import com.google.gson.Gson;
+import com.qmth.themis.business.config.EnvConfig;
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.entity.TBTaskHistory;
+import com.qmth.themis.business.service.TBTaskHistoryService;
+import com.qmth.themis.business.util.OssUtil;
+
+import javax.swing.*;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+import java.util.StringJoiner;
+
+/**
+ * @Description: 任务公用
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/19
+ */
+public class TaskImportCommon {
+    private Integer mode = null;
+    private Map<String, Object> map = null;
+    private Long orgId = null;
+    private Long createId = null;
+    private Long examId = null;
+    private TBTaskHistoryService tbTaskHistoryService;
+    private String type;
+    private String path;
+    private Integer uploadType;
+    private OssUtil ossUtil;
+    private EnvConfig envConfig;
+    private Map<String, Object> ossEnv;
+    private TBTaskHistory tbTaskHistory;
+
+    public TaskImportCommon(Map<String, Object> map) {
+        this.map = map;
+    }
+
+    public void init() {
+        Gson gson = new Gson();
+        this.mode = Integer.parseInt(String.valueOf(this.map.get("mode")));
+        this.orgId = Long.parseLong(String.valueOf(this.map.get("orgId")));
+        this.createId = Long.parseLong(String.valueOf(this.map.get("createId")));
+        this.examId = Long.parseLong(String.valueOf(this.map.get("examId")));
+        JSONObject remarkJs = JSONObject.parseObject(String.valueOf(this.map.get("remark")));
+        this.type = String.valueOf(remarkJs.get("type"));
+        this.path = String.valueOf(remarkJs.get("path"));
+        this.uploadType = Integer.parseInt(String.valueOf(remarkJs.get("uploadType")));
+        this.ossUtil = SpringContextHolder.getBean(OssUtil.class);
+        this.envConfig = SpringContextHolder.getBean(EnvConfig.class);
+        this.ossEnv = this.envConfig.getOssEnv(this.uploadType);
+        Map tbTaskHistoryMap = (Map) this.map.get("tbTaskHistory");
+        tbTaskHistoryMap = SystemConstant.timeTransform(tbTaskHistoryMap);
+        this.tbTaskHistory = gson.fromJson(gson.toJson(tbTaskHistoryMap), TBTaskHistory.class);
+        this.tbTaskHistoryService = SpringContextHolder.getBean(TBTaskHistoryService.class);
+    }
+
+    /**
+     * 获取上传的文件
+     *
+     * @return
+     */
+    public File getUploadFile() {
+        StringJoiner localPath = new StringJoiner("").add(
+                System.getProperty(SystemConstant.USER_DIR)).add(File.separator).add(path);
+        File file = null;
+        if (Objects.nonNull(this.type) && Objects.equals(this.type, SystemConstant.LOCAL)) {
+            file = new File(localPath.toString());
+        } else {
+            try {
+                file = this.ossUtil.ossDownload(this.ossEnv, localPath.toString(), localPath.toString());
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+        return file;
+    }
+
+    /**
+     * 写入txt文件
+     *
+     * @param exception
+     * @param txtStr
+     * @throws IOException
+     */
+    public void writeImportResultTxt(boolean exception, String txtStr) throws IOException {
+        FileWriter fileWriter = null;
+        try {
+            this.path = this.path.substring(0, this.path.lastIndexOf(File.separator) + 1);
+            File file = new File(this.path + this.tbTaskHistory.getId() + ".txt");
+            if (!file.exists()) {
+                file.createNewFile();
+            }
+            fileWriter = new FileWriter(file);
+            fileWriter.write(txtStr);
+            JSONObject json = new JSONObject();
+            json.put("path", file.getPath());
+            json.put("type", this.type);
+            if (exception) {
+                this.tbTaskHistory.setErrorFilePath(json.toJSONString());
+            } else {
+                this.tbTaskHistory.setResultFilePath(json.toJSONString());
+            }
+            if (Objects.equals(this.type, SystemConstant.OSS)) {
+                this.ossUtil.ossUpload(this.ossEnv, file.getPath(), txtStr);
+                file.delete();
+            }
+            tbTaskHistoryService.updateById(this.tbTaskHistory);
+        } catch (IOException e) {
+            e.printStackTrace();
+        } finally {
+            if (Objects.nonNull(fileWriter)) {
+                fileWriter.flush();
+                fileWriter.close();
+            }
+        }
+    }
+
+    public Integer getMode() {
+        return mode;
+    }
+
+    public Map<String, Object> getMap() {
+        return map;
+    }
+
+    public Long getOrgId() {
+        return orgId;
+    }
+
+    public Long getCreateId() {
+        return createId;
+    }
+
+    public Long getExamId() {
+        return examId;
+    }
+
+    public TBTaskHistoryService getTbTaskHistoryService() {
+        return tbTaskHistoryService;
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public String getPath() {
+        return path;
+    }
+
+    public Integer getUploadType() {
+        return uploadType;
+    }
+
+    public OssUtil getOssUtil() {
+        return ossUtil;
+    }
+
+    public EnvConfig getEnvConfig() {
+        return envConfig;
+    }
+
+    public Map<String, Object> getOssEnv() {
+        return ossEnv;
+    }
+
+    public TBTaskHistory getTbTaskHistory() {
+        return tbTaskHistory;
+    }
+}

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

@@ -0,0 +1,239 @@
+package com.qmth.themis.business.templete.impl;
+
+import cn.hutool.core.date.DateUtil;
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.google.common.collect.Lists;
+import com.google.gson.Gson;
+import com.qmth.themis.business.constant.SpringContextHolder;
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.dto.ExamStudentDtoImport;
+import com.qmth.themis.business.entity.*;
+import com.qmth.themis.business.enums.ExamModeEnum;
+import com.qmth.themis.business.enums.RoleEnum;
+import com.qmth.themis.business.enums.TaskTypeExecEnum;
+import com.qmth.themis.business.service.*;
+import com.qmth.themis.business.templete.TaskImportCommon;
+import com.qmth.themis.business.templete.TaskImportTemplete;
+import com.qmth.themis.business.util.ExcelCallback;
+import com.qmth.themis.business.util.ExcelError;
+import com.qmth.themis.business.util.ExcelUtil;
+import com.qmth.themis.business.util.JacksonUtil;
+import com.qmth.themis.common.exception.BusinessException;
+import com.qmth.themis.common.util.Result;
+import com.qmth.themis.common.util.ResultUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.LinkedMultiValueMap;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * @Description: 考生导入任务
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/15
+ */
+public class TaskExamStudentImportTemplete implements TaskImportTemplete {
+    private final static Logger log = LoggerFactory.getLogger(TaskExamStudentImportTemplete.class);
+
+    /**
+     * 考生导入模版
+     *
+     * @param map
+     * @return
+     * @throws IOException
+     * @throws InterruptedException
+     */
+    @Override
+    @Transactional
+    public Result importTask(Map<String, Object> map) throws IOException {
+        TaskImportCommon taskImportCommon = new TaskImportCommon(map);
+        taskImportCommon.init();
+        File file = taskImportCommon.getUploadFile();
+        List<LinkedMultiValueMap<Integer, Object>> finalList = ExcelUtil.excelReader(new FileInputStream(file), Lists.newArrayList(ExamStudentDtoImport.class), new ExcelCallback() {
+            @Override
+            public List<LinkedMultiValueMap<Integer, Object>> callback(List<LinkedMultiValueMap<Integer, Object>> finalList, List<LinkedMultiValueMap<Integer, String>> finalColumnNameList) throws IllegalAccessException {
+                List<ExcelError> excelErrorList = new ArrayList<>();
+                for (int i = 0; i < finalList.size(); i++) {
+                    LinkedMultiValueMap<Integer, Object> map = finalList.get(i);
+                    List<Object> examStudentDtoImportList = map.get(i);
+                    for (int y = 0; y < examStudentDtoImportList.size(); y++) {
+                        ExamStudentDtoImport examStudentDtoImport = (ExamStudentDtoImport) examStudentDtoImportList.get(y);
+                        List<ExcelError> excelErrorTemp = ExcelUtil.checkExcelField(examStudentDtoImport, y, i);
+                        if (excelErrorTemp.size() > 0) {
+                            excelErrorList.addAll(excelErrorTemp);
+                        }
+                        //校验考试模式，如果是集中统一，则需填写考试场次
+                        if (taskImportCommon.getMode().intValue() == ExamModeEnum.together.ordinal() && Objects.isNull(examStudentDtoImport.getExamActivityCode())) {
+                            excelErrorList.add(new ExcelError(y + 1, "excel第" + (i + 1) + "个sheet第" + (y + 1) + "行考场为空"));
+                        }
+                    }
+                }
+                if (excelErrorList.size() > 0) {
+                    throw new BusinessException(JSONObject.toJSONString(excelErrorList));
+                }
+                return finalList;
+            }
+        });
+
+        //保存数据到数据库
+        if (Objects.nonNull(finalList) && finalList.size() > 0) {
+            log.info("开始导入考生数据");
+            long start = System.currentTimeMillis();
+            TBTaskHistoryService tbTaskHistoryService = taskImportCommon.getTbTaskHistoryService();
+            TBTaskHistory tbTaskHistory = taskImportCommon.getTbTaskHistory();
+
+            TEExamStudentService teExamStudentService = SpringContextHolder.getBean(TEExamStudentService.class);
+            TEStudentService teStudentService = SpringContextHolder.getBean(TEStudentService.class);
+            TBUserRoleService tbUserRoleService = SpringContextHolder.getBean(TBUserRoleService.class);
+            TBExamInvigilateUserService tbExamInvigilateUserService = SpringContextHolder.getBean(TBExamInvigilateUserService.class);
+            boolean exception = false;
+            String timeFormat = "yyyy-MM-dd HH:mm:ss";
+
+            List<TEExamStudent> teExamStudentList = new ArrayList<>();
+            List<TEStudent> teStudentList = new ArrayList<>();
+            List<TBUserRole> tbUserRoleList = new ArrayList<>();
+            List<TBExamInvigilateUser> tbExamInvigilateUserList = new ArrayList<>();
+            List txtList = new ArrayList();
+            txtList.add(DateUtil.format(new Date(), timeFormat) + "->开始准备处理导入的考生数据");
+            int y = 0, min = 0;
+            try {
+                Gson gson = new Gson();
+                Set<String> roomCodeAndNameSet = new HashSet<>();
+                Map<String, TEExamActivity> teExamActivityMap = (Map<String, TEExamActivity>) map.get("teExamActivityMap");
+                //保存用户
+                for (int i = 0; i < finalList.size(); i++) {
+                    LinkedMultiValueMap<Integer, Object> finalMap = finalList.get(i);
+                    List<Object> examStudentDtoImportList = finalMap.get(i);
+                    min = 0;
+                    int max = SystemConstant.MAX_IMPORT_SIZE, size = examStudentDtoImportList.size();
+                    if (max >= size) {
+                        max = size;
+                    }
+                    while (max <= size) {
+                        List subList = examStudentDtoImportList.subList(min, max);
+                        for (y = 0; y < subList.size(); y++) {
+                            ExamStudentDtoImport examStudentDtoImport = (ExamStudentDtoImport) subList.get(y);
+                            Map m = (Map) teExamActivityMap.get(examStudentDtoImport.getExamActivityCode());
+                            m = SystemConstant.timeTransform(m);
+                            TEExamActivity teExamActivity = gson.fromJson(gson.toJson(m), TEExamActivity.class);
+                            //先根据证件号+科目代码查询考生是否存在，存在则更新，不存在则插入
+                            QueryWrapper<TEExamStudent> teExamStudentQueryWrapper = new QueryWrapper<>();
+                            teExamStudentQueryWrapper.lambda().eq(TEExamStudent::getIdentity, examStudentDtoImport.getIdentity()).eq(TEExamStudent::getCourseCode, examStudentDtoImport.getCourseCode());
+                            TEExamStudent teExamStudent = teExamStudentService.getOne(teExamStudentQueryWrapper);
+                            //如果为空则插入考生数据，插入考生前先插入学生档案数据
+                            if (Objects.isNull(teExamStudent)) {
+                                //先插入学生档案数据
+                                TEStudent teStudent = new TEStudent(taskImportCommon.getOrgId(), examStudentDtoImport.getIdentity(), examStudentDtoImport.getName(), taskImportCommon.getCreateId());
+                                teStudentService.save(teStudent);
+                                teStudentList.add(teStudent);
+
+                                //插入用户角色关系
+                                TBUserRole tbUserRole = new TBUserRole(teStudent.getId(), RoleEnum.STUDENT.name());
+                                tbUserRoleService.save(tbUserRole);
+                                tbUserRoleList.add(tbUserRole);
+
+                                teExamStudent = gson.fromJson(gson.toJson(examStudentDtoImport), TEExamStudent.class);
+                                teExamStudent.setExamId(taskImportCommon.getExamId());
+                                teExamStudent.setExamActivityId(teExamActivity.getId());
+                                teExamStudent.setStudentId(teStudent.getId());
+                                teExamStudent.setCreateId(taskImportCommon.getCreateId());
+                            } else {
+                                teExamStudent.setUpdateId(taskImportCommon.getCreateId());
+                                teExamStudent.setName(examStudentDtoImport.getName());
+                                teExamStudent.setCourseName(examStudentDtoImport.getCourseName());
+                                teExamStudent.setGrade(examStudentDtoImport.getGrade());
+                                teExamStudent.setClassNo(examStudentDtoImport.getClassNo());
+                                teExamStudent.setRoomCode(examStudentDtoImport.getRoomCode());
+                                teExamStudent.setRoomName(examStudentDtoImport.getRoomName());
+                                teExamStudent.setExamActivityId(teExamActivity.getId());
+                            }
+                            teExamStudentService.saveOrUpdate(teExamStudent);
+                            roomCodeAndNameSet.add(teExamStudent.getRoomCode() + ":" + teExamStudent.getRoomName());
+                            teExamStudentList.add(teExamStudent);
+                        }
+                        BigDecimal bigDecimal = new BigDecimal(100);
+                        BigDecimal progress = new BigDecimal(Double.valueOf(new BigDecimal(max).divide(new BigDecimal(size), 2, BigDecimal.ROUND_HALF_UP).multiply(bigDecimal).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue())).setScale(0, BigDecimal.ROUND_HALF_UP);
+                        log.info("progress:{}", progress);
+                        if (progress.intValue() == 100) {
+                            tbTaskHistory.setStatus(TaskTypeExecEnum.finish.ordinal());
+                            tbTaskHistory.setSummary("共处理了" + size + "条数据");
+                            tbTaskHistory.setFinishTime(new Date());
+                            txtList.add(DateUtil.format(new Date(), timeFormat) + "->考生数据处理完毕，" + tbTaskHistory.getSummary());
+                        } else {
+                            tbTaskHistory.setStatus(TaskTypeExecEnum.running.ordinal());
+                            tbTaskHistory.setSummary("正在处理第" + min + "条至" + max + "条数据");
+                            txtList.add(DateUtil.format(new Date(), timeFormat) + "->考生数据处理中，" + tbTaskHistory.getSummary());
+                        }
+                        tbTaskHistory.setProgress(progress.doubleValue());
+                        tbTaskHistoryService.updateById(tbTaskHistory);
+                        if (max == size) {
+                            break;
+                        }
+                        min = max;
+                        max += 500;
+                        if (max >= size) {
+                            max = size;
+                        }
+                    }
+                }
+                //考场创建
+                if (Objects.nonNull(roomCodeAndNameSet) && roomCodeAndNameSet.size() > 0) {
+                    Map<String, String> tbExamInvigilateUserMap = (Map<String, String>) map.get("tbExamInvigilateUserMap");
+                    AtomicInteger count = new AtomicInteger(0);
+                    roomCodeAndNameSet.forEach(s -> {
+                        if (Objects.isNull(tbExamInvigilateUserMap) || (Objects.nonNull(tbExamInvigilateUserMap) && Objects.isNull(tbExamInvigilateUserMap.get(s)))) {
+                            String[] strs = s.split(":");
+                            TBExamInvigilateUser tbExamInvigilateUser = new TBExamInvigilateUser(taskImportCommon.getOrgId(), taskImportCommon.getCreateId(), strs[0], strs[1]);
+                            tbExamInvigilateUserService.save(tbExamInvigilateUser);
+                            tbExamInvigilateUserList.add(tbExamInvigilateUser);
+                            count.getAndIncrement();
+                        }
+                    });
+                    if (count.get() > 0) {
+                        txtList.add(DateUtil.format(new Date(), timeFormat) + "->创建了" + count + "条考场数据");
+                    }
+                }
+                long end = System.currentTimeMillis();
+                log.info("导入考生数据结束,============耗时============:{}秒", (end - start) / 1000);
+            } catch (Exception e) {
+                e.printStackTrace();
+                exception = true;
+                String exceptionStr = "考生数据处理到第" + (min + y + 1) + "条时发生异常:" + e.getMessage();
+                txtList.add(DateUtil.format(new Date(), timeFormat) + "->" + exceptionStr);
+                tbTaskHistory.setSummary(exceptionStr);
+                tbTaskHistoryService.updateById(tbTaskHistory);
+                if (Objects.nonNull(teStudentList) && teStudentList.size() > 0) {
+                    teStudentService.removeByIds(teStudentList.stream().map(s -> s.getId()).collect(Collectors.toList()));
+                }
+                if (Objects.nonNull(tbUserRoleList) && tbUserRoleList.size() > 0) {
+                    tbUserRoleService.removeByIds(tbUserRoleList.stream().map(s -> s.getId()).collect(Collectors.toList()));
+                }
+                if (Objects.nonNull(teExamStudentList) && teExamStudentList.size() > 0) {
+                    teExamStudentService.removeByIds(teExamStudentList.stream().map(s -> s.getId()).collect(Collectors.toList()));
+                }
+                if (Objects.nonNull(tbExamInvigilateUserList) && tbExamInvigilateUserList.size() > 0) {
+                    tbExamInvigilateUserService.removeByIds(tbExamInvigilateUserList.stream().map(s -> s.getId()).collect(Collectors.toList()));
+                }
+                if (e instanceof BusinessException) {
+                    throw new BusinessException(e.getMessage());
+                } else {
+                    throw new RuntimeException(e);
+                }
+            } finally {
+                //这里写入txt文件
+                taskImportCommon.writeImportResultTxt(exception, txtList.toString());
+            }
+        }
+        return ResultUtil.ok(JacksonUtil.parseJson(SystemConstant.SUCCESS));
+    }
+}

themis-business/src/main/java/com/qmth/themis/business/templete/impl/TaskExamStudentTemplete.java → themis-business/src/main/java/com/qmth/themis/business/templete/impl/TaskRoomCodeImportTemplete.java

@@ -40,8 +40,8 @@ import java.util.stream.Collectors;
  * @Author: wangliang
  * @Date: 2020/7/15
  */
-public class TaskExamStudentTemplete implements TaskImportTemplete {
-    private final static Logger log = LoggerFactory.getLogger(TaskExamStudentTemplete.class);
+public class TaskRoomCodeImportTemplete implements TaskImportTemplete {
+    private final static Logger log = LoggerFactory.getLogger(TaskRoomCodeImportTemplete.class);
 
     /**
      * 考生导入模版

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

@@ -5,7 +5,8 @@ import com.qmth.themis.business.constant.SystemConstant;
 import com.qmth.themis.business.entity.TMRocketMessage;
 import com.qmth.themis.business.service.TMRocketMessageService;
 import com.qmth.themis.business.templete.TaskImportTemplete;
-import com.qmth.themis.business.templete.impl.TaskExamStudentTemplete;
+import com.qmth.themis.business.templete.impl.TaskExamStudentImportTemplete;
+import com.qmth.themis.business.templete.impl.TaskRoomCodeImportTemplete;
 import com.qmth.themis.business.threadPool.MyThreadPool;
 import com.qmth.themis.business.util.JacksonUtil;
 import com.qmth.themis.business.util.RedisUtil;
@@ -41,7 +42,7 @@ import java.util.concurrent.TimeUnit;
  * @Date: 2020/7/2
  */
 @Service
-public class RocketTaskConsumer implements MessageListenerConcurrently {
+public class RocketTaskConsumer {
     private final static Logger log = LoggerFactory.getLogger(RocketTaskConsumer.class);
 
     @Resource
@@ -53,54 +54,78 @@ public class RocketTaskConsumer implements MessageListenerConcurrently {
     @Resource
     MyThreadPool myThreadPool;
 
-    @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-        MqDto mqDto = null;
-        try {
-            long threadId = Thread.currentThread().getId();
-            String threadName = Thread.currentThread().getName();
-            Gson gson = new Gson();
-            for (MessageExt messageExt : msgs) {
-                log.info(":{}-:{} taskConsumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
-//                MqDto mqDto = (MqDto) toJavaObject(parseObject(new String(messageExt.getBody(), Constants.CHARSET)), MqDto.class);
-                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
-                log.info(":{}-:{} taskConsumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
-                log.info(":{}-:{} taskConsumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
-                if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.getMqTopicList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
-                    Map<String, Object> map = (Map<String, Object>) mqDto.getBody();
-                    myThreadPool.arbitratePoolTaskExecutor.execute(() -> {
-                        TaskImportTemplete examStudentTemplete = new TaskExamStudentTemplete();
-                        try {
-                            examStudentTemplete.importTask(map);
-                        } catch (IOException e) {
-                            e.printStackTrace();
+    /**
+     * 考生导入
+     */
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.taskConsumerExamStudentImportGroup}", topic = "${mq.config.taskTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.taskTopicExamStudentImportTag}")
+    public class taskConsumerExamStudent implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+
+        @Override
+        public void onMessage(Message message) {
+            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
+        }
+
+        @Override
+        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
+            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
+            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
+            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
+
+                @Override
+                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+                    MqDto mqDto = null;
+                    try {
+                        long threadId = Thread.currentThread().getId();
+                        String threadName = Thread.currentThread().getName();
+                        Gson gson = new Gson();
+                        for (MessageExt messageExt : msgs) {
+                            log.info(":{}-:{} task ExamStudentImport Consumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+                            mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+                            log.info(":{}-:{} task ExamStudentImport Consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
+                            log.info(":{}-:{} task ExamStudentImport Consumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
+                            if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.getMqTopicList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
+                                Map<String, Object> map = (Map<String, Object>) mqDto.getBody();
+                                myThreadPool.arbitratePoolTaskExecutor.execute(() -> {
+                                    TaskImportTemplete examStudentImportTemplete = new TaskExamStudentImportTemplete();
+                                    try {
+                                        examStudentImportTemplete.importTask(map);
+                                    } catch (IOException e) {
+                                        e.printStackTrace();
+                                    }
+                                });
+                                mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
+                                mqDto.setBody(JacksonUtil.parseJson(mqDto.getBody()));
+                                TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+                                tmRocketMessageService.saveOrUpdate(tmRocketMessage);
+                                redisUtil.deleteMqTopicList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId());
+                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                            } else {
+                                log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
+                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                            }
                         }
-                    });
-                    mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
-                    mqDto.setBody(JacksonUtil.parseJson(mqDto.getBody()));
-                    TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
-                    tmRocketMessageService.saveOrUpdate(tmRocketMessage);
-                    redisUtil.deleteMqTopicList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId());
-                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                } else {
-                    log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
-                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                    } finally {
+                        if (Objects.nonNull(mqDto)) {
+                            redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId());
+                        }
+                    }
+                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
                 }
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
-        } finally {
-            if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId());
-            }
+            });
         }
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
     }
 
+    /**
+     * 考场导入
+     */
     @Service
-    @RocketMQMessageListener(consumerGroup = "${mq.config.taskConsumerExamStudentGroup}", topic = "${mq.config.taskTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.taskTopicExamStudentTag}")
-    public class taskConsumerExamStudent implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+    @RocketMQMessageListener(consumerGroup = "${mq.config.taskConsumerRoomCodeImportGroup}", topic = "${mq.config.taskTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.taskTopicRoomCodeImportTag}")
+    public class taskConsumerRoomCodeImport implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
 
         @Override
         public void onMessage(Message message) {
@@ -112,25 +137,52 @@ public class RocketTaskConsumer implements MessageListenerConcurrently {
             defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
             defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
             defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
-            defaultMQPushConsumer.registerMessageListener(RocketTaskConsumer.this::consumeMessage);
+            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
+
+                @Override
+                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+                    MqDto mqDto = null;
+                    try {
+                        long threadId = Thread.currentThread().getId();
+                        String threadName = Thread.currentThread().getName();
+                        Gson gson = new Gson();
+                        for (MessageExt messageExt : msgs) {
+                            log.info(":{}-:{} task roomCodeImport Consumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
+                            mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
+                            log.info(":{}-:{} task roomCodeImport Consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
+                            log.info(":{}-:{} task roomCodeImport Consumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
+                            if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.getMqTopicList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
+                                Map<String, Object> map = (Map<String, Object>) mqDto.getBody();
+                                myThreadPool.arbitratePoolTaskExecutor.execute(() -> {
+                                    TaskImportTemplete roomCodeImportTemplete = new TaskRoomCodeImportTemplete();
+                                    try {
+                                        roomCodeImportTemplete.importTask(map);
+                                    } catch (IOException e) {
+                                        e.printStackTrace();
+                                    }
+                                });
+                                mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
+                                mqDto.setBody(JacksonUtil.parseJson(mqDto.getBody()));
+                                TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+                                tmRocketMessageService.saveOrUpdate(tmRocketMessage);
+                                redisUtil.deleteMqTopicList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId());
+                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                            } else {
+                                log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
+                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                            }
+                        }
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                    } finally {
+                        if (Objects.nonNull(mqDto)) {
+                            redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId());
+                        }
+                    }
+                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+                }
+            });
         }
     }
-
-//    @Service
-//    @RocketMQMessageListener(consumerGroup = "${mq.config.userLogConsumerStudentGroup}", topic = "${mq.config.userLogTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.userLogTopicStudentTag}")
-//    public class sessionConsumerStudentLog implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
-//
-//        @Override
-//        public void onMessage(Message message) {
-//            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
-//        }
-//
-//        @Override
-//        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
-//            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
-//            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
-//            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
-//            defaultMQPushConsumer.registerMessageListener(RocketTaskConsumer.this::consumeMessage);
-//        }
-//    }
-}
+}

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

@@ -203,5 +203,8 @@ mq.config.userLogConsumerStudentGroup=${mq.config.userLogConsumerGroup}-${mq.con
 mq.config.taskTopic=${mq.config.server}-topic-task
 mq.config.taskConsumerGroup=${mq.config.server}-group-task
 
-mq.config.taskTopicExamStudentTag=examStudent
-mq.config.taskConsumerExamStudentGroup=${mq.config.taskConsumerGroup}-${mq.config.taskTopicExamStudentTag}
+mq.config.taskTopicExamStudentImportTag=examStudentImport
+mq.config.taskConsumerExamStudentImportGroup=${mq.config.taskConsumerGroup}-${mq.config.taskTopicExamStudentImportTag}
+
+mq.config.taskTopicRoomCodeImportTag=roomCodeImport
+mq.config.taskConsumerRoomCodeImportGroup=${mq.config.taskConsumerGroup}-${mq.config.taskTopicRoomCodeImportTag}