package com.qmth.themis.mq.service.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.google.gson.Gson;
import com.qmth.themis.business.cache.ExamActivityRecordCacheUtil;
import com.qmth.themis.business.cache.ExamRecordCacheUtil;
import com.qmth.themis.business.cache.RedisKeyHelper;
import com.qmth.themis.business.cache.bean.ExamActivityRecordCacheBean;
import com.qmth.themis.business.cache.bean.ExamCacheBean;
import com.qmth.themis.business.cache.bean.ExamStudentCacheBean;
import com.qmth.themis.business.constant.SpringContextHolder;
import com.qmth.themis.business.constant.SystemConstant;
import com.qmth.themis.business.dto.MqDto;
import com.qmth.themis.business.dto.WarningDto;
import com.qmth.themis.business.entity.*;
import com.qmth.themis.business.enums.*;
import com.qmth.themis.business.service.*;
import com.qmth.themis.business.templete.TaskExportTemplete;
import com.qmth.themis.business.templete.TaskImportTemplete;
import com.qmth.themis.business.templete.impl.*;
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.util.SimpleBeanUtil;
import com.qmth.themis.mq.service.MqLogicService;
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 javax.annotation.Resource;
import java.io.IOException;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;

/**
 * @Description: mq执行逻辑 impl
 * @Param:
 * @return:
 * @Author: wangliang
 * @Date: 2020/7/31
 */
@Service
public class MqLogicServiceImpl implements MqLogicService {
    private final static Logger log = LoggerFactory.getLogger(MqLogicServiceImpl.class);

    @Resource
    RedisUtil redisUtil;

    @Resource
    TMRocketMessageService tmRocketMessageService;

    @Resource
    TEUserLogService teUserLogService;

    @Resource
    TEExamStudentLogService teExamStudentLogService;

    @Resource
    TBSessionService tbSessionService;

    @Resource
    MyThreadPool myThreadPool;

    @Resource
    TOeExamBreakHistoryService tOeExamBreakHistoryService;

    @Resource
    TOeExamRecordService examRecordService;

    @Resource
    TOeFaceVerifyHistoryService faceVerifyHistoryService;

    @Resource
    TOeLivenessVerifyHistoryService livenessVerifyHistoryService;

    @Resource
    CommonService commonService;

    @Resource
    TIeExamInvigilateCallLogService tIeExamInvigilateCallLogService;

    @Resource
    TIeExamInvigilateCallService tIeExamInvigilateCallService;

    @Resource
    TIeInvigilateExceptionInfoService tIeInvigilateExceptionInfoService;

    @Resource
    TEExamService teExamService;

    @Resource
    TEExamStudentService teExamStudentService;

    @Resource
    WarningService warningService;

    @Resource
    MqDtoService mqDtoService;

    @Resource
    TEConfigService teConfigService;

    /**
     * mq最大重试次数逻辑
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqMaxReconsumeTime(MqDto mqDto, String key) {
        //超过最大重试次数，保存到数据库，后续可以发短信通知系统管理人员
        Gson gson = new Gson();
        mqDto.setAck(SystemConstant.POSION_ACK_TYPE);
        TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
        tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
        tmRocketMessageService.saveOrUpdate(tmRocketMessage);
        redisUtil.delete(key, mqDto.getId());
    }

    /**
     * 用户轨迹逻辑
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqUserLogLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        String tag = mqDto.getTag();
        if (Objects.equals(MqTagEnum.USER.name(), tag)) {
            TEUserLog teUserLog = new TEUserLog(String.valueOf(mqDto.getBody()), SystemOperationEnum.valueOf(String.valueOf(mqDto.getBody())).getCode(), SystemOperationEnum.valueOf(String.valueOf(mqDto.getBody())).getCode(), Long.parseLong(String.valueOf(mqDto.getObjId())));
            teUserLogService.save(teUserLog);
        } else if (Objects.equals(MqTagEnum.STUDENT.name(), tag)) {
            TEExamStudentLog teExamStudentLog = null;
            if (Objects.nonNull(mqDto.getProperties())) {
                teExamStudentLog = new TEExamStudentLog(String.valueOf(mqDto.getBody()), SystemOperationEnum.valueOf(String.valueOf(mqDto.getBody())).getCode(), String.valueOf(mqDto.getProperties().get("remark")), Long.parseLong(String.valueOf(mqDto.getObjId())), Long.parseLong(String.valueOf(mqDto.getProperties().get("examStudentId"))), Long.parseLong(String.valueOf(mqDto.getProperties().get("examRecordId"))));
            } else {
                teExamStudentLog = new TEExamStudentLog(String.valueOf(mqDto.getBody()), SystemOperationEnum.valueOf(String.valueOf(mqDto.getBody())).getCode(), SystemOperationEnum.valueOf(String.valueOf(mqDto.getBody())).getCode(), Long.parseLong(String.valueOf(mqDto.getObjId())));
            }
            teExamStudentLogService.save(teExamStudentLog);
        }
        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(key, mqDto.getId());
    }

    /**
     * session逻辑
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqSessionLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        tbSessionService.saveSessionInfo(JacksonUtil.readJson(JacksonUtil.parseJson(mqDto.getBody()), TBSession.class), mqDto.getTimestamp());
        mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
        TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
        tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
        tmRocketMessageService.saveOrUpdate(tmRocketMessage);
        redisUtil.delete(key, mqDto.getId());
    }

    /**
     * task逻辑
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqTaskLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        Map<String, Object> map = (Map<String, Object>) mqDto.getBody();
        String tag = mqDto.getTag();
        myThreadPool.arbitratePoolTaskExecutor.execute(() -> {
            if (tag.contains("Import".toUpperCase())) {
                TaskImportTemplete taskImportTemplete = null;
                if (Objects.equals(MqTagEnum.EXAM_STUDENT_IMPORT.name(), tag)) {
                    taskImportTemplete = SpringContextHolder.getBean(TaskExamStudentImportTemplete.class);
                } else if (Objects.equals(MqTagEnum.ROOM_CODE_IMPORT.name(), tag)) {
                    taskImportTemplete = SpringContextHolder.getBean(TaskRoomCodeImportTemplete.class);
                } else if (Objects.equals(MqTagEnum.EXAM_PAPER_IMPORT.name(), tag)) {
                    taskImportTemplete = SpringContextHolder.getBean(TaskExamPaperImportTemplete.class);
                }
                try {
                    taskImportTemplete.importTask(map);
                } catch (IOException e) {
                    log.error("请求出错", e);
                }
            } else {
                TaskExportTemplete taskExportTemplete = null;
                if (Objects.equals(MqTagEnum.ROOM_CODE_EXPORT.name(), tag)) {
                    taskExportTemplete = SpringContextHolder.getBean(TaskRoomCodeExportTemplete.class);
                } else if (Objects.equals(MqTagEnum.EXAM_STUDENT_EXPORT.name(), tag)) {
                    taskExportTemplete = SpringContextHolder.getBean(TaskExamStudentExportTemplete.class);
                }
                try {
                    taskExportTemplete.exportTask(map);
                } catch (IOException e) {
                    log.error("请求出错", e);
                }
            }
        });
        mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
        mqDto.setBody(JacksonUtil.parseJson(mqDto.getBody()));
        TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
        tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
        tmRocketMessageService.saveOrUpdate(tmRocketMessage);
        redisUtil.delete(key, mqDto.getId());
    }

    /**
     * websocket超时退出逻辑
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqWebsocketUnNormalLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);//表示成功处理
        Map<String, Object> tranMap = mqDto.getProperties();
        Long recordId = Long.parseLong(String.valueOf(tranMap.get("recordId")));
        ExamRecordStatusEnum status = ExamRecordCacheUtil.getStatus(recordId);
        WebsocketStatusEnum websocketStatusEnum = ExamRecordCacheUtil.getClientWebsocketStatus(recordId);
        if (Objects.nonNull(websocketStatusEnum) && !Objects.equals(websocketStatusEnum, WebsocketStatusEnum.ON_LINE) && (!Objects.equals(status, ExamRecordStatusEnum.FIRST_PREPARE) || !Objects.equals(status, ExamRecordStatusEnum.FINISHED) || !Objects.equals(status, ExamRecordStatusEnum.PERSISTED))) {
            Long examId = ExamRecordCacheUtil.getExamId(recordId);
            Long examActivityId = ExamRecordCacheUtil.getExamActivityId(recordId);
            Long examStudentId = ExamRecordCacheUtil.getExamStudentId(recordId);
            ExamStudentCacheBean examStudentCacheBean = teExamStudentService.getExamStudentCacheBean(examStudentId);
            ExamCacheBean ec = teExamService.getExamCacheBean(examId);//考试缓存
            Long breakId = null;
            //增加断点记录,获取剩余断点次数
            Integer alreadyBreakCount = Objects.isNull(ExamRecordCacheUtil.getAlreadyBreakCount(recordId)) ? 0 : ExamRecordCacheUtil.getAlreadyBreakCount(recordId);
            Integer leftBreakResumeCount = ec.getBreakResumeCount() - alreadyBreakCount;
            if (leftBreakResumeCount < 0) {
                Integer durationSeconds = Objects.isNull(ExamRecordCacheUtil.getDurationSeconds(recordId)) ? 0 : ExamRecordCacheUtil.getDurationSeconds(recordId);
                teExamService.finish(examStudentCacheBean.getStudentId(), recordId, FinishTypeEnum.AUTO.name(), durationSeconds);
            } else {
                alreadyBreakCount++;
                breakId = Constants.idGen.next();
                ExamRecordCacheUtil.setLastBreakId(recordId, breakId, false);
                ExamRecordCacheUtil.setStatus(recordId, ExamRecordStatusEnum.BREAK_OFF, false);
                Long lastBreakTime = System.currentTimeMillis();
                ExamRecordCacheUtil.setLastBreakTime(recordId, lastBreakTime, false);
                ExamRecordCacheUtil.setAlreadyBreakCount(recordId, alreadyBreakCount, false);
                Long lastStartTime = System.currentTimeMillis();
                ExamRecordCacheUtil.setLastStartTime(recordId, lastStartTime, false);
                String[] columns = new String[]{ExamRecordFieldEnum.last_break_id.name(), ExamRecordFieldEnum.status.name(), ExamRecordFieldEnum.last_break_time.name(), ExamRecordFieldEnum.already_break_count.name(), ExamRecordFieldEnum.last_start_time.name()};
                Object[] values = new Object[]{breakId, ExamRecordStatusEnum.BREAK_OFF, lastBreakTime, alreadyBreakCount, lastStartTime};
                examRecordService.dataUpdatesMq(recordId, columns, values);
                //考试断点异常原因 发送mq start
                MqDto mqDtoBreak = new MqDto(MqTopicEnum.THEMIS_TOPIC.getCode(), MqTagEnum.EXAM_BREAK.name(), ExceptionEnum.NET_TIME_OUT, MqTagEnum.EXAM_BREAK, String.valueOf(recordId), String.valueOf(recordId));
                mqDtoService.assembleSendOneWayMsg(mqDtoBreak);
                //考试断点异常原因 发送mq end
                //更新场次-考试记录缓存
                ExamActivityRecordCacheUtil.setExamRecordStatus(examActivityId, recordId, new ExamActivityRecordCacheBean(ExamRecordCacheUtil.getExamStudentId(recordId), ExamRecordCacheUtil.getStatus(recordId)));
            }
            //发送移动端监考退出考试mq消息 start
            MqDto mqDtoExamStop = new MqDto(MqTopicEnum.THEMIS_TOPIC.getCode(), MqTagEnum.EXAM_STOP.name(), recordId, MqTagEnum.EXAM_STOP, String.valueOf(recordId), String.valueOf(recordId));
            mqDtoService.assembleSendOneWayMsg(mqDtoExamStop);
            //发送移动端监考退出考试mq消息 end
        }
        TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
        tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
        tmRocketMessageService.saveOrUpdate(tmRocketMessage);
        redisUtil.delete(key, mqDto.getId());
    }

    /**
     * 计算客观分逻辑
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqCalculateObjectiveScoreLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        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(key, mqDto.getId());
    }

    /**
     * 人脸验证保存
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqFaceVerifySaveLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
        Long id = (Long) param.get("id");
        Long recordId = (Long) param.get("recordId");
        String type = (String) param.get("type");
        String photoUrl = (String) param.get("photoUrl");
        Integer faceCount = (Integer) param.get("faceCount");
        Double similarity = (Double) param.get("similarity");
        Double realness = (Double) param.get("realness");
        Long time = (Long) param.get("time");
        String exception = (String) param.get("exception");
        TEConfig teConfig = teConfigService.getGlobalConfig();
        if (Objects.equals(type.toUpperCase(), ExamTypeEnum.IN_PROCESS.name())) {
            VerifyExceptionEnum warningEnum = VerifyExceptionEnum.valueOf(exception);
            WarningDto warningDto = new WarningDto(warningEnum, faceCount, realness, recordId, photoUrl);
            Long examStudentId = ExamRecordCacheUtil.getExamStudentId(recordId);
            ExamStudentCacheBean examStudentCacheBean = teExamStudentService.getExamStudentCacheBean(examStudentId);
            if (Objects.equals(VerifyExceptionEnum.NONE, warningEnum)) {//无异常，往考生日志表里插一条
                JSONObject jsonObject = new JSONObject();
                jsonObject.put(PhotoTypeEnum.FACE_VERIFY_PHOTO.name(), photoUrl);
                TEExamStudentLog teExamStudentLog = new TEExamStudentLog(type, ExamTypeEnum.valueOf(type).getCode(), jsonObject.toJSONString(), examStudentCacheBean.getStudentId(), examStudentCacheBean.getId(), recordId);
                faceVerifyHistoryService.save(id, recordId, type, photoUrl, faceCount, similarity, realness, time, exception, null);
                teExamStudentLogService.saveOrUpdate(teExamStudentLog);
            } else {
                if (Objects.equals(VerifyExceptionEnum.FACE_COUNT_ERROR, warningEnum)) {//人脸数量异常
                    if (faceCount > 1) {//多张人脸
                        String multipleFaceRandom = ExamRecordCacheUtil.getMultipleFaceCountErrorRandom(recordId);
                        if (Objects.isNull(multipleFaceRandom)) {
                            multipleFaceRandom = String.valueOf(UUID.randomUUID()).replaceAll("-", "");
                            ExamRecordCacheUtil.setMultipleFaceCountErrorRandom(recordId, multipleFaceRandom);
                        }
                        ExamRecordCacheUtil.setMultipleFaceCountErrorNum(recordId, ExamRecordCacheUtil.getMultipleFaceCountErrorNum(recordId) + 1);
                        warningDto.setRandom(multipleFaceRandom);
                        faceVerifyHistoryService.save(id, recordId, type, photoUrl, faceCount, similarity, realness, time, exception, multipleFaceRandom);
                        Integer count = ExamRecordCacheUtil.getMultipleFaceCountErrorNum(recordId);
                        if (count >= teConfig.getMultipleFaceCountError()) {
                            ExamRecordCacheUtil.setMultipleFaceCountErrorRandom(recordId, String.valueOf(UUID.randomUUID()).replaceAll("-", ""));
                            ExamRecordCacheUtil.setMultipleFaceCountErrorNum(recordId, 0);
                            warningService.faceCountError(warningDto);
                        }
                    } else if (faceCount <= 0) {
                        String noFaceRandom = ExamRecordCacheUtil.getNoFaceCountErrorRandom(recordId);
                        if (Objects.isNull(noFaceRandom)) {
                            noFaceRandom = String.valueOf(UUID.randomUUID()).replaceAll("-", "");
                            ExamRecordCacheUtil.setNoFaceCountErrorRandom(recordId, noFaceRandom);
                        }
                        ExamRecordCacheUtil.setNoFaceCountErrorNum(recordId, ExamRecordCacheUtil.getNoFaceCountErrorNum(recordId) + 1);
                        warningDto.setRandom(noFaceRandom);
                        faceVerifyHistoryService.save(id, recordId, type, photoUrl, faceCount, similarity, realness, time, exception, noFaceRandom);
                        Integer count = ExamRecordCacheUtil.getNoFaceCountErrorNum(recordId);
                        if (count >= teConfig.getNoFaceCountError()) {
                            ExamRecordCacheUtil.setNoFaceCountErrorRandom(recordId, String.valueOf(UUID.randomUUID()).replaceAll("-", ""));
                            ExamRecordCacheUtil.setNoFaceCountErrorNum(recordId, 0);
                            warningService.faceCountError(warningDto);
                        }
                    }
                } else if (Objects.equals(VerifyExceptionEnum.FACE_COMPARE_ERROR, warningEnum)) {//人脸比对异常
                    String faceCompareErrorRandom = ExamRecordCacheUtil.getFaceCompareErrorRandom(recordId);
                    if (Objects.isNull(faceCompareErrorRandom)) {
                        faceCompareErrorRandom = String.valueOf(UUID.randomUUID()).replaceAll("-", "");
                        ExamRecordCacheUtil.setFaceCompareErrorRandom(recordId, faceCompareErrorRandom);
                    }
                    ExamRecordCacheUtil.setFaceCompareErrorNum(recordId, ExamRecordCacheUtil.getFaceCompareErrorNum(recordId) + 1);
                    warningDto.setRandom(faceCompareErrorRandom);
                    faceVerifyHistoryService.save(id, recordId, type, photoUrl, faceCount, similarity, realness, time, exception, null);
                    Integer count = ExamRecordCacheUtil.getFaceCompareErrorNum(recordId);
                    if (count >= teConfig.getTotalFaceCompareErrorCount()) {
                        ExamRecordCacheUtil.setFaceCompareErrorRandom(recordId, String.valueOf(UUID.randomUUID()).replaceAll("-", ""));
                        ExamRecordCacheUtil.setFaceCompareErrorNum(recordId, 0);
                        warningService.faceCountError(warningDto);
                    }
                } else {
                    faceVerifyHistoryService.save(id, recordId, type, photoUrl, faceCount, similarity, realness, time, exception, null);
                    if (realness.intValue() == 0) {//真实性异常
                        warningService.realnessError(warningDto);
                    }
                }
            }
        } else {
            faceVerifyHistoryService.save(id, recordId, type, photoUrl, faceCount, similarity, realness, time, exception, null);
        }
        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(key, mqDto.getId());
    }

    /**
     * 活体验证
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqLivenessVerifySaveLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
        Long id = (Long) param.get("id");
        Long recordId = (Long) param.get("recordId");
        String type = (String) param.get("type");
        String actions = (String) param.get("actions");
        Integer retry = (Integer) param.get("retry");
        Long startTime = (Long) param.get("startTime");
        Long finishTime = (Long) param.get("finishTime");
        String exception = (String) param.get("exception");
        TEConfig teConfig = teConfigService.getGlobalConfig();
        if (Objects.equals(type.toUpperCase(), ExamTypeEnum.IN_PROCESS.name())) {
            VerifyExceptionEnum warningEnum = VerifyExceptionEnum.valueOf(exception);
            JSONArray jsonArray = JSONArray.parseArray(actions);
            Long examStudentId = ExamRecordCacheUtil.getExamStudentId(recordId);
            ExamStudentCacheBean examStudentCacheBean = teExamStudentService.getExamStudentCacheBean(examStudentId);
            for (int i = 0; i < jsonArray.size(); i++) {
                JSONObject jsonObject = jsonArray.getJSONObject(i);
                if (Objects.equals(VerifyExceptionEnum.NONE, warningEnum)) {//无异常，往考生日志表里插一条
                    JSONObject object = new JSONObject();
                    object.put(PhotoTypeEnum.LIVENESS_VERIFY_PHOTO.name(), String.valueOf(jsonObject.get("photoUrl")));
                    TEExamStudentLog teExamStudentLog = new TEExamStudentLog(type, LivenessTypeEnum.valueOf(type).getTitle(), object.toJSONString(), examStudentCacheBean.getStudentId(), examStudentCacheBean.getId(), recordId);
                    livenessVerifyHistoryService.save(id, recordId, type, actions, retry, startTime, finishTime, exception, null);
                    teExamStudentLogService.saveOrUpdate(teExamStudentLog);
                } else {
                    if (Objects.equals(VerifyExceptionEnum.FACE_COUNT_ERROR, warningEnum)) {//人脸数量异常
                        Integer faceCount = Integer.parseInt(String.valueOf(jsonObject.get("faceCount")));
                        WarningDto warningDto = new WarningDto(warningEnum, faceCount, null, recordId, String.valueOf(jsonObject.get("photoUrl")));
                        if (faceCount > 1) {//多张人脸
                            String multipleFaceRandom = ExamRecordCacheUtil.getMultipleFaceCountErrorRandom(recordId);
                            if (Objects.isNull(multipleFaceRandom)) {
                                multipleFaceRandom = String.valueOf(UUID.randomUUID()).replaceAll("-", "");
                                ExamRecordCacheUtil.setMultipleFaceCountErrorRandom(recordId, multipleFaceRandom);
                            }
                            ExamRecordCacheUtil.setMultipleFaceCountErrorNum(recordId, ExamRecordCacheUtil.getMultipleFaceCountErrorNum(recordId) + 1);
                            warningDto.setRandom(multipleFaceRandom);
                            livenessVerifyHistoryService.save(id, recordId, type, actions, retry, startTime, finishTime, exception, multipleFaceRandom);
                            Integer count = ExamRecordCacheUtil.getMultipleFaceCountErrorNum(recordId);
                            if (count >= teConfig.getMultipleFaceCountError()) {
                                ExamRecordCacheUtil.setMultipleFaceCountErrorRandom(recordId, String.valueOf(UUID.randomUUID()).replaceAll("-", ""));
                                ExamRecordCacheUtil.setMultipleFaceCountErrorNum(recordId, 0);
                                warningService.faceCountError(warningDto);
                            }
                        } else if (faceCount <= 0) {
                            String noFaceRandom = ExamRecordCacheUtil.getNoFaceCountErrorRandom(recordId);
                            if (Objects.isNull(noFaceRandom)) {
                                noFaceRandom = String.valueOf(UUID.randomUUID()).replaceAll("-", "");
                                ExamRecordCacheUtil.setNoFaceCountErrorRandom(recordId, noFaceRandom);
                            }
                            ExamRecordCacheUtil.setNoFaceCountErrorNum(recordId, ExamRecordCacheUtil.getMultipleFaceCountErrorNum(recordId) + 1);
                            warningDto.setRandom(noFaceRandom);
                            livenessVerifyHistoryService.save(id, recordId, type, actions, retry, startTime, finishTime, exception, noFaceRandom);
                            Integer count = ExamRecordCacheUtil.getNoFaceCountErrorNum(recordId);
                            if (count >= teConfig.getNoFaceCountError()) {
                                ExamRecordCacheUtil.setNoFaceCountErrorRandom(recordId, String.valueOf(UUID.randomUUID()).replaceAll("-", ""));
                                ExamRecordCacheUtil.setNoFaceCountErrorNum(recordId, 0);
                                warningService.faceCountError(warningDto);
                            }
                        }
                    } else if (Objects.equals(VerifyExceptionEnum.FACE_COMPARE_ERROR, warningEnum)) {//人脸比对异常
                        WarningDto warningDto = new WarningDto(warningEnum, null, null, recordId, String.valueOf(jsonObject.get("photoUrl")));
                        String faceCompareErrorRandom = ExamRecordCacheUtil.getFaceCompareErrorRandom(recordId);
                        if (Objects.isNull(faceCompareErrorRandom)) {
                            faceCompareErrorRandom = String.valueOf(UUID.randomUUID()).replaceAll("-", "");
                            ExamRecordCacheUtil.setFaceCompareErrorRandom(recordId, faceCompareErrorRandom);
                        }
                        ExamRecordCacheUtil.setFaceCompareErrorNum(recordId, ExamRecordCacheUtil.getFaceCompareErrorNum(recordId) + 1);
                        warningDto.setRandom(faceCompareErrorRandom);
                        livenessVerifyHistoryService.save(id, recordId, type, actions, retry, startTime, finishTime, exception, faceCompareErrorRandom);
                        Integer count = ExamRecordCacheUtil.getFaceCompareErrorNum(recordId);
                        if (count >= teConfig.getTotalFaceCompareErrorCount()) {
                            ExamRecordCacheUtil.setFaceCompareErrorRandom(recordId, String.valueOf(UUID.randomUUID()).replaceAll("-", ""));
                            ExamRecordCacheUtil.setFaceCompareErrorNum(recordId, 0);
                            warningService.faceCountError(warningDto);
                        }
                    }
                }
            }
        } else {
            livenessVerifyHistoryService.save(id, recordId, type, actions, retry, startTime, finishTime, exception, null);
        }
        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(key, mqDto.getId());
    }

    /**
     * 考试记录数据持久化
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqRecordPersistedLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
        Long recordId = (Long) param.get("recordId");
        commonService.persisted(recordId);
        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(key, mqDto.getId());
    }

    @Override
    @Transactional
    public void execMqRecordUpdateLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        String tag = mqDto.getTag();
        if (Objects.equals(tag, MqTagEnum.EXAM_RECORD_UPDATE.name())) {//单字段更新
            Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
            Long recordId = Long.parseLong(String.valueOf(param.get("recordId")));
            String colName = (String) param.get("colName");
            Object colValue = param.get("colValue");
            examRecordService.dataUpdate(recordId, colName, colValue);
        } else if (Objects.equals(tag, MqTagEnum.EXAM_RECORD_UPDATE_COLUMNS.name())) {//多字段更新
            Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
            Long recordId = Long.parseLong(String.valueOf(param.get("recordId")));
            List<String> colNames = (List<String>) (param.get("colNames"));
            List<Object> colValues = (List<Object>) param.get("colValues");
            UpdateWrapper<TOeExamRecord> tOeExamRecordUpdateWrapper = new UpdateWrapper<>();
            for (int i = 0; i < colNames.size(); i++) {
                String colName = colNames.get(i);
                Object colValue = colValues.get(i);
                tOeExamRecordUpdateWrapper.set(colName, colValue);
            }
            tOeExamRecordUpdateWrapper.lambda().eq(TOeExamRecord::getId, recordId);
            examRecordService.update(tOeExamRecordUpdateWrapper);
        }
        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(key, mqDto.getId());
    }

    @Override
    @Transactional
    public void execMqRecordInitLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
        examRecordService.dataInit(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(key, mqDto.getId());
    }

    /**
     * 监控通话申请逻辑
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqLogLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        String tag = mqDto.getTag();
        if (Objects.equals(MqTagEnum.MONITOR_LOG.name(), tag)) {//监考日志
            TIeExamInvigilateCallLog tIeExamInvigilateCallLog = JacksonUtil.readJson(JacksonUtil.parseJson(mqDto.getBody()), TIeExamInvigilateCallLog.class);
            Long recordId = tIeExamInvigilateCallLog.getExamRecordId();
            Long examId = null, examActivityId = null, examStudentId = null;
            if (Objects.nonNull(ExamRecordCacheUtil.getExamActivityId(recordId))) {
                examActivityId = ExamRecordCacheUtil.getExamActivityId(recordId);
                examId = ExamRecordCacheUtil.getExamId(recordId);
                examStudentId = ExamRecordCacheUtil.getExamStudentId(recordId);
            } else {
                TOeExamRecord tOeExamRecord = examRecordService.getById(tIeExamInvigilateCallLog.getExamRecordId());
                examActivityId = tOeExamRecord.getExamActivityId();
                examId = tOeExamRecord.getExamId();
                examStudentId = tOeExamRecord.getExamStudentId();
            }
            tIeExamInvigilateCallLog.setExamId(examId);
            tIeExamInvigilateCallLog.setExamActivityId(examActivityId);
            tIeExamInvigilateCallLog.setExamStudentId(examStudentId);

            QueryWrapper<TIeExamInvigilateCall> tIeExamInvigilateCallQueryWrapper = new QueryWrapper<>();
            tIeExamInvigilateCallQueryWrapper.lambda().eq(TIeExamInvigilateCall::getExamRecordId, tIeExamInvigilateCallLog.getExamRecordId()).eq(TIeExamInvigilateCall::getSource, tIeExamInvigilateCallLog.getSource());
            TIeExamInvigilateCall tIeExamInvigilateCall = tIeExamInvigilateCallService.getOne(tIeExamInvigilateCallQueryWrapper);
            if (Objects.isNull(tIeExamInvigilateCall)) {
                tIeExamInvigilateCall = gson.fromJson(gson.toJson(tIeExamInvigilateCallLog), TIeExamInvigilateCall.class);
                tIeExamInvigilateCall.setId(Constants.idGen.next());
            } else {
                tIeExamInvigilateCall.setLiveUrl(tIeExamInvigilateCallLog.getLiveUrl());
                tIeExamInvigilateCall.setStatus(tIeExamInvigilateCallLog.getStatus());
                tIeExamInvigilateCall.setCallStatus(tIeExamInvigilateCallLog.getCallStatus());
            }
            if (Objects.nonNull(tIeExamInvigilateCall.getUpdateTime())) {
                Long dbtimestamp = Objects.nonNull(tIeExamInvigilateCall.getUpdateTime()) ? tIeExamInvigilateCall.getUpdateTime() : 0L;
                if (mqDto.getTimestamp() > dbtimestamp) {
                    tIeExamInvigilateCallService.updateById(tIeExamInvigilateCall);
                }
            } else {
                tIeExamInvigilateCallService.saveOrUpdate(tIeExamInvigilateCall);
            }
            tIeExamInvigilateCallLogService.saveOrUpdate(tIeExamInvigilateCallLog);
        } else if (Objects.equals(MqTagEnum.EXCEPTION_LOG.name(), tag)) {//考试断点异常日志
            JSONObject jsonObject = JSONObject.parseObject(String.valueOf(mqDto.getBody()));
            ExceptionEnum exceptionEnum = ExceptionEnum.valueOf(ExceptionEnum.convertToName(String.valueOf(jsonObject.getJSONObject("reason").get("type"))));
            Long recordId = Long.parseLong(mqDto.getObjId());
            Long breakId = ExamRecordCacheUtil.getLastBreakId(recordId);
            if (Objects.nonNull(breakId)) {
                //获取断点时间
                Long lastBreakTime = ExamRecordCacheUtil.getLastBreakTime(recordId);
                Integer diff = 0;
                if (Objects.nonNull(lastBreakTime)) {
                    Long l = ((System.currentTimeMillis() - lastBreakTime) / 1000);
                    diff = l.intValue();
                }

                //修改异常日志
                String reason = String.valueOf(jsonObject.getJSONObject("reason").get("reason"));
                UpdateWrapper<TIeInvigilateExceptionInfo> tIeInvigilateExceptionInfoUpdateWrapper = new UpdateWrapper<>();
                tIeInvigilateExceptionInfoUpdateWrapper.lambda().set(TIeInvigilateExceptionInfo::getInfo, reason)
                        .set(TIeInvigilateExceptionInfo::getType, exceptionEnum)
                        .set(TIeInvigilateExceptionInfo::getDifference, diff)
                        .set(TIeInvigilateExceptionInfo::getUpdateTime, System.currentTimeMillis())
                        .eq(TIeInvigilateExceptionInfo::getObjId, breakId);
                tIeInvigilateExceptionInfoService.update(tIeInvigilateExceptionInfoUpdateWrapper);

                //修改考生轨迹
                UpdateWrapper<TEExamStudentLog> teExamStudentLogUpdateWrapper = new UpdateWrapper<>();
                teExamStudentLogUpdateWrapper.lambda().set(TEExamStudentLog::getType, exceptionEnum.name())
                        .set(TEExamStudentLog::getInfo, exceptionEnum.getCode())
                        .set(TEExamStudentLog::getRemark, reason)
                        .set(TEExamStudentLog::getUpdateTime, System.currentTimeMillis())
                        .eq(TEExamStudentLog::getObjId, breakId);
                teExamStudentLogService.update(teExamStudentLogUpdateWrapper);
            }
        } else if (Objects.equals(MqTagEnum.WARNING_LOG.name(), tag)) {//考试预警日志
            //todo 预警先预留
        }
        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(key, mqDto.getId());
    }

    /**
     * 重新算分
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqCalculateScoreLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
        Long examId = (Long) param.get("examId");
        Long taskId = (Long) param.get("taskId");
        teExamService.calculateScore(examId, taskId);
        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(key, mqDto.getId());
    }

    /**
     * 更新考生信息
     */
    @Override
    @Transactional
    public void execMqExamStudentUpdateLogic(MqDto mqDto, String key) {
        Gson gson = new Gson();
        Map<String, Object> param = (Map<String, Object>) mqDto.getBody();
        teExamStudentService.updateByMqMsg(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(key, mqDto.getId());
    }

    /**
     * 考试断点
     *
     * @param mqDto
     * @param key
     */
    @Override
    @Transactional
    public void execMqExamBreakLogic(MqDto mqDto, String key) {
        ExceptionEnum exceptionEnum = ExceptionEnum.valueOf(String.valueOf(mqDto.getBody()));
        Long recordId = Long.parseLong(mqDto.getObjId());

        //先查询之前的断点记录
        QueryWrapper<TOeExamBreakHistory> tOeExamBreakHistoryQueryWrapper = new QueryWrapper<>();
        tOeExamBreakHistoryQueryWrapper.lambda().eq(TOeExamBreakHistory::getExamRecordId, recordId);
        List<TOeExamBreakHistory> tOeExamBreakHistoryList = tOeExamBreakHistoryService.list(tOeExamBreakHistoryQueryWrapper);
        //删除历史断点缓存
        if (Objects.nonNull(tOeExamBreakHistoryList) && tOeExamBreakHistoryList.size() > 0) {
            tOeExamBreakHistoryList.forEach(s -> {
                redisUtil.delete(RedisKeyHelper.examBreakCacheKey(s.getId()));
            });
        }
        //增加断点记录
        TOeExamBreakHistory tOeExamBreakHistory = new TOeExamBreakHistory(ExamRecordCacheUtil.getLastBreakId(recordId), recordId, System.currentTimeMillis(), exceptionEnum, exceptionEnum.name());
        tOeExamBreakHistoryService.saveOrUpdate(tOeExamBreakHistory);
        redisUtil.setForHash(RedisKeyHelper.examBreakCacheKey(tOeExamBreakHistory.getId()), SimpleBeanUtil.objectToMap(tOeExamBreakHistory));

        Long clientLastSyncTime = ExamRecordCacheUtil.getClientLastSyncTime(recordId);
        Integer diff = 0;
        if (Objects.nonNull(clientLastSyncTime)) {//大于等于当前时间，说明未重连或重登录
            Long l = (System.currentTimeMillis() - clientLastSyncTime) / 1000;
            diff = l.intValue();
        }
        Long examId = ExamRecordCacheUtil.getExamId(recordId);
        Long examActivityId = ExamRecordCacheUtil.getExamActivityId(recordId);
        Long examStudentId = ExamRecordCacheUtil.getExamStudentId(recordId);
        ExamStudentCacheBean examStudentCacheBean = teExamStudentService.getExamStudentCacheBean(examStudentId);
        //增加异常日志
        TIeInvigilateExceptionInfo tIeInvigilateExceptionInfo = new TIeInvigilateExceptionInfo(examId, examActivityId, recordId, examStudentId, ExceptionEnum.NET_TIME_OUT.getCode(), ExceptionEnum.NET_TIME_OUT, diff, ExamRecordCacheUtil.getLastBreakId(recordId));
        tIeInvigilateExceptionInfoService.saveOrUpdate(tIeInvigilateExceptionInfo);
        TEExamStudentLog teExamStudentLog = new TEExamStudentLog(SystemOperationEnum.BREAK_OFF.name(), SystemOperationEnum.BREAK_OFF.getCode(), SystemOperationEnum.BREAK_OFF.getCode(), examStudentCacheBean.getStudentId(), examStudentId, recordId, ExamRecordCacheUtil.getLastBreakId(recordId));
        teExamStudentLogService.save(teExamStudentLog);

        Gson gson = new Gson();
        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(key, mqDto.getId());
    }

    /**
     * 普通消息
     *
     * @param msgs
     * @param consumeConcurrentlyContext
     * @param mqExecTypeEnum
     * @return
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext, MqExecTypeEnum mqExecTypeEnum) {
        MqDto mqDto = null;
        try {
            long threadId = Thread.currentThread().getId();
            String threadName = Thread.currentThread().getName();
            for (MessageExt messageExt : msgs) {
                log.info(":{}-:{} consumer 重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
                log.info(":{}-:{} consumer 接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
                int reconsumeTime = messageExt.getReconsumeTimes();
                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
                    this.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
                } 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)) {
                        Method method = this.getClass().getDeclaredMethod(mqExecTypeEnum.getDesc(), MqDto.class, String.class);
                        log.info(":{}-:{} 准备执行mq exec:{}逻辑", threadId, threadName, method.getName());
                        method.invoke(this, mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                }
            }
        } catch (Exception e) {
            log.error("mq普通消息监听，消息消费出错", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
        } finally {
            if (Objects.nonNull(mqDto)) {
                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
    }

    /**
     * 延时消息
     *
     * @param msgs
     * @param consumeConcurrentlyContext
     * @param mqExecTypeEnum
     * @return
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessageDelay(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext, MqExecTypeEnum mqExecTypeEnum) {
        MqDto mqDto = null;
        try {
            long threadId = Thread.currentThread().getId();
            String threadName = Thread.currentThread().getName();
            for (MessageExt messageExt : msgs) {
                log.info(":{}-:{} consumer重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
                mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
                log.info(":{}-:{} consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
                int reconsumeTime = messageExt.getReconsumeTimes();
                if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
                    this.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_DELAY_TOPIC_BUFFER_LIST);
                } else {
                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(SystemConstant.MQ_DELAY_TOPIC_BUFFER_LIST) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
                        Method method = this.getClass().getDeclaredMethod(mqExecTypeEnum.getDesc(), MqDto.class, String.class);
                        log.info(":{}-:{} 准备执行mq exec:{}逻辑", threadId, threadName, method.getName());
                        method.invoke(this, mqDto, SystemConstant.MQ_DELAY_TOPIC_BUFFER_LIST);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                }
            }
        } catch (Exception e) {
            log.error("mq延时消息监听，消息消费出错", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
        } finally {
            if (Objects.nonNull(mqDto)) {
                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId());
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
    }
}
