处理长期任务发送问题

liusheng

5 天以前 0bbbd746e8c39d91ad7b47a4e36e29ab972dc98c

 ruoyi-quartz/src/main/java/com/ruoyi/quartz/service/impl/CollectHISServiceImpl.java

@@ -1,9 +1,13 @@
package com.ruoyi.quartz.service.impl;

import com.ruoyi.othervisit.api.OtherVisitFacade;
import com.ruoyi.inhospvisit.api.InhospVisitFacade;
import com.google.gson.Gson;
import com.ruoyi.common.core.domain.entity.SysDept;
import com.ruoyi.common.core.domain.entity.SysUser;
import com.ruoyi.common.core.domain.entity.SysUserDept;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.utils.ArrayListUtils;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.quartz.mapper.CollectHISMapper;
import com.ruoyi.quartz.service.ICollectHISService;
@@ -26,11 +30,26 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;

import com.ruoyi.othervisit.domain.PatMedOperationItem;
import com.ruoyi.outpatientvisit.mapper.PatMedOuthospMapper;
import com.ruoyi.othervisit.domain.PatMedOperation;
import com.ruoyi.outpatientvisit.domain.PatMedOuthosp;
import com.ruoyi.inhospvisit.domain.PatMedInhosp;
import com.ruoyi.common.core.domain.entity.PatArchive;
import com.ruoyi.task.domain.entity.ServiceSubtaskEntity;
import com.ruoyi.task.domain.ServiceSubtask;
import com.ruoyi.task.mapper.ServiceSubtaskMapper;
@Slf4j
@Service
public class CollectHISServiceImpl implements ICollectHISService {
    @Autowired
    private CollectHISMapper chMapper;

    /**
     * 手术数据一律经 ruoyi-othervisit 的门面取用（跨模块数据访问原则）。
     */
    @Autowired
    private OtherVisitFacade otherVisitFacade;

    @Autowired
    private SysUser2Mapper sysUser2Mapper;
@@ -54,10 +73,13 @@
    private PatMedOuthospMapper patMedOuthospMapper;

    @Autowired
    private PatMedInhospMapper patMedInhospMapper;
    private InhospVisitFacade inhospVisitFacade;

    @Autowired
    private PatArchiveMapper patArchiveMapper;

    @Autowired
    private ServiceSubtaskMapper serviceSubtaskMapper;

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;
@@ -68,17 +90,16 @@
    @Autowired
    private UtilsMapper utilsMapper;

    @Autowired
    private RedisCache redisCache;

    private static final long SHARDING_THRESHOLD = 5_000_000L;
    private static final String PAT_ARCHIVE_TABLE = "pat_archive";
    private static final String PAT_MED_INHOSP_TABLE = "pat_med_inhosp";

    @Value("${spring.profiles.active}")
    private String active;
    @Autowired
    private PatMedOperationMapper patMedOperationMapper;

    @Autowired
    private PatMedOperationItemMapper patMedOperationItemMapper;

    @Override
    public List<PatArchive> selectPatArchiveList(PatArchive patArchive) {
@@ -108,7 +129,7 @@
        Integer count = 0;
        List<Map<String, Object>> syncList = utilsMapper.getList("SELECT id,syncName,DATE_FORMAT(ADDDATE(startTime, INTERVAL -1 MINUTE), '%Y-%m-%d %H:%i:%s') startTime  FROM sys_sync_time  where state=0");
        if (syncList.size() > 0) {
            String endTime = chMapper.getSqlString("select sysdate from healthy_inhosp where rownum=1");
            String endTime = chMapper.getSqlString("select (sysdate-1) as endtime from healthy_inhosp where rownum=1");
            //同步患者基本信息
//            List<Map<String, Object>> syncTemp = syncList.stream().filter(row -> row.get("syncName").equals("pat_archive")).collect(Collectors.toList());
//            if (syncTemp.size() > 0) {
@@ -229,7 +250,7 @@
                //同步出入院
                PatMedInhosp pmi = new PatMedInhosp();
                pmi.setSerialnum(pm.getSerialnum());
                List<PatMedInhosp> patMedInhospList1 = patMedInhospMapper.selectPatMedInhospListBySerialnum(pmi);
                List<PatMedInhosp> patMedInhospList1 = inhospVisitFacade.selectPatMedInhospListBySerialnum(pmi);
                if (patMedInhospList1.size() == 0) {
                    pm.setCreateTime(new Date());
                    //市一 院区id对应orgid，机构id为默认的1
@@ -237,7 +258,7 @@
                        pm.setCampusid(pm.getOrgid());
                        pm.setOrgid("1");
                    }
                    count += patMedInhospMapper.insertPatMedInhosp(pm);
                    count += inhospVisitFacade.insertPatMedInhosp(pm);
                } else {
                    pm.setInhospid(patMedInhospList1.get(0).getInhospid());
                    pm.setUpdateTime(new Date());
@@ -246,7 +267,7 @@
                        pm.setCampusid(pm.getOrgid());
                        pm.setOrgid("1");
                    }
                    count += patMedInhospMapper.updatePatMedInhosp(pm);
                    count += inhospVisitFacade.updatePatMedInhosp(pm);
                }
            } catch (Exception ex) {
                log.info("同步出入院:" + ex.getMessage());
@@ -339,6 +360,11 @@
        log.info("同步手术信息数量" + patMedOperations.size());
        for (PatMedOperation operation : patMedOperations) {
            try {
                if(ObjectUtils.isNotEmpty(operation.getCanceltime())){
                    //作废删除操作
                    cancelOper(operation);
                    continue;
                }
                Long patId = patIdHisMap.get(operation.getPatidHis());
                if (patId == null) {
                    patId = syncSinglePatient(operation.getPatidHis());
@@ -351,16 +377,16 @@
                    operation.setPatid(patId);
                }
                PatMedOperation oper = new PatMedOperation();
                oper.setOpid(operation.getOpid());
                List<PatMedOperation> patMedOperations1 = patMedOperationMapper.selectPatMedOperationList(oper);
                oper.setOpreqid(operation.getOpreqid());
                List<PatMedOperation> patMedOperations1 = otherVisitFacade.selectOperationList(oper);
                if (CollectionUtils.isNotEmpty(patMedOperations1)) {
                    operation.setId(patMedOperations1.get(0).getId());
                    operation.setUpdateTime(new Date());
                    count += patMedOperationMapper.updatePatMedOperation(patMedOperations1.get(0));
                    count += otherVisitFacade.updateOperation(operation);
                    syncOperItem(operation);
                } else {
                    operation.setCreateTime(new Date());
                    count += patMedOperationMapper.insertPatMedOperation(operation);
                    count += otherVisitFacade.insertOperation(operation);
                    syncOperItem(operation);
                }
            } catch (Exception ex) {
@@ -370,6 +396,59 @@
        return count;
    }

    private Integer cancelOper(PatMedOperation oper){
        PatMedOperation operationVo = new PatMedOperation();
        operationVo.setOpreqid(oper.getOpreqid());
        List<PatMedOperation> operations = otherVisitFacade.selectOperationList(operationVo);
        PatMedOperation operation = new PatMedOperation();
        if(CollectionUtils.isEmpty(operations)){
            return 0;
        }else {
            operation = operations.get(0);
        }
        log.info("手术作废id=" + operation.getId());
        PatMedOperationItem pmoi = new PatMedOperationItem();
        //获取需要出院手术随访，未处理的数据
        pmoi.setOpid(operation.getId().toString());
        List<PatMedOperationItem> patMedOperationItems
                = otherVisitFacade.selectOperationItemListByCondition(pmoi);
        List<Long> operationItemIds = patMedOperationItems.stream()
                .map(PatMedOperationItem::getId)
                .filter(Objects::nonNull)
                .distinct()
                .collect(Collectors.toList());

        //不需要服务了，需要通过patid和sendstate=1,2,3,5,7去查询一下，service_subtask里是否有正在执行的任务，
        //有的话，立即停了
        List<Long> sendstates = Arrays.asList(1L, 2L, 3L, 5L, 7L);
        List<ServiceSubtask> serviceSubtaskList = new ArrayList<>();
        ServiceSubtaskEntity serviceSubtaskVO = new ServiceSubtaskEntity();
        serviceSubtaskVO.setOrgid(operation.getOrgid());
        serviceSubtaskVO.setPatid(operation.getPatid());
        serviceSubtaskVO.setSendstates(sendstates);
        serviceSubtaskVO.setOperationItemIds(operationItemIds);
        serviceSubtaskList = serviceSubtaskMapper.selectServiceSubtaskList(serviceSubtaskVO);
        //去redis中，查询是否有subid，有的话移除cache-exist
        if (CollectionUtils.isNotEmpty(serviceSubtaskList)) {
            for (ServiceSubtask serviceSubtask : serviceSubtaskList) {
                log.info("需要移除的subId为：{}", serviceSubtask.getId().toString());
                serviceSubtask.setSendstate(4L);
                serviceSubtask.setRemark("手术撤销");
                serviceSubtaskMapper.updateServiceSubtask(serviceSubtask);
                redisCache.removeElementFromList("cache-exist", serviceSubtask.getId().toString());
                redisCache.removeElementFromList("cache-0", serviceSubtask.getId().toString());
                redisCache.removeElementFromList("cache-1", serviceSubtask.getId().toString());
                redisCache.removeElementFromList("cache-2", serviceSubtask.getId().toString());
                redisCache.removeElementFromList("cache-3", serviceSubtask.getId().toString());
                redisCache.removeElementFromList("cache-4", serviceSubtask.getId().toString());
            }
        }
        operation.setUpdateTime(new Date());
        otherVisitFacade.updateOperation(operation);
        otherVisitFacade.deleteSoftOperationItemByOpId(operation.getOpid());
        return 1;
    }

    /**
     * 同步手术信息
     *
@@ -377,37 +456,72 @@
     * @return
     */
    private Integer syncOperItem(PatMedOperation operation) {
        if(ObjectUtils.isNotEmpty(operation.getId())){
            operation.setOpid(operation.getId().toString());
        }
        operation.setUpdateTime(new Date());
        otherVisitFacade.updateOperation(operation);

        Integer count = 0;
        PatMedOperationItem patMedOperationItem = new PatMedOperationItem();
        patMedOperationItem.setOrgid(operation.getOrgid());
        if(StringUtils.isNotEmpty(operation.getOpid())){
            patMedOperationItem.setOpid(Long.valueOf(operation.getOpid()));
            patMedOperationItem.setLastStartTime(operation.getLastStartTime());
            patMedOperationItem.setLastEndTime(operation.getLastEndTime());
            List<PatMedOperationItem> patMedOperationItems = chMapper.selectOperItemList(patMedOperationItem);
            PatMedOperationItem patMedOperationItem = new PatMedOperationItem();
            //patMedOperation.opReqid = patMedOperationItem.operationid
            List<PatMedOperationItem> patMedOperationItems = new ArrayList<>();
            if (ObjectUtils.isNotEmpty(operation.getOpreqid())){
                patMedOperationItem.setOperationid(operation.getOpreqid());
                patMedOperationItems = chMapper.selectOperItemList(patMedOperationItem);
            }else {
                log.info("同步手术详情信息:手术申请单ID为空!");
                return count;
            }

            PatMedOperationItem operItem = new PatMedOperationItem();
            operItem.setOperationid(operation.getOpreqid());
            List<PatMedOperationItem> patMedOperationItems1 = otherVisitFacade.selectOperationItemList(operItem);

            Map<String, List<PatMedOperationItem>> maps = ArrayListUtils.splitListsById(patMedOperationItems,
                    patMedOperationItems1, PatMedOperationItem::getOperationid);
            //只在视图的数据不在本地的数据，执行新增操作
            List<PatMedOperationItem> addItems = maps.get("onlyIn1");
            //只在本地的数据不在视图的数据，执行删除操作
            List<PatMedOperationItem> delItems = maps.get("onlyIn2");
            //重叠的数据，执行更新操作 patMedOperationItems的数据
            List<PatMedOperationItem> editItems = maps.get("common");

            log.info("同步手术详情信息数量" + patMedOperationItems.size());
            for (PatMedOperationItem operationItem : patMedOperationItems) {
                try {
                    PatMedOperationItem operItem = new PatMedOperationItem();
                    operItem.setOpid(operationItem.getOpid());
                    operItem.setOperationid(operationItem.getOperationid());
                    List<PatMedOperationItem> patMedOperationItems1 = patMedOperationItemMapper.selectPatMedOperationItemList(operItem);
                    if (CollectionUtils.isNotEmpty(patMedOperationItems1)) {
                        operationItem.setId(patMedOperationItems1.get(0).getId());
                        operationItem.setUpdateTime(new Date());
                        count += patMedOperationItemMapper.updatePatMedOperationItem(operationItem);
                    } else {
                        operation.setCreateTime(new Date());
                        count += patMedOperationItemMapper.insertPatMedOperationItem(operationItem);
                    }
                } catch (Exception ex) {
                    log.info("同步手术详情信息:" + ex.getMessage());

            try{
                for (PatMedOperationItem operationItem : addItems) {
                    operationItem.setInhospstate(operation.getInhospstate());
                    operationItem.setSerialnum(operation.getSerialnum());
                    operationItem.setOpid(operation.getId().toString());
                    operationItem.setOperationid(operation.getOpreqid());
                    operationItem.setOpercheckFlag("0");
                    operationItem.setCreateTime(new Date());
                    count += otherVisitFacade.insertOperationItem(operationItem);
                }

                for (PatMedOperationItem operationItem : delItems) {
                    operationItem.setUpdateTime(new Date());
                    count += otherVisitFacade.deleteOperationItemByOperationid(operationItem.getOperationid());
                }

                for (PatMedOperationItem operationItem : editItems) {
                    operationItem.setInhospstate(operation.getInhospstate());
                    operationItem.setSerialnum(operation.getSerialnum());
                    operationItem.setOpid(operation.getId().toString());
                    operationItem.setOperationid(operation.getOpreqid());
                    operationItem.setUpdateTime(new Date());
                    count += otherVisitFacade.updateOperationItemByOperationid(operationItem);
                }
                log.info("同步手术id={}详情信息数量, 新增{}条，删除{}条，修改{}条", operation.getId(), addItems.size(), delItems.size(), editItems.size());
            } catch (Exception ex) {
                log.info("同步手术详情信息:" + ex.getMessage());
            }

//            try {
//                patMedOperationItemMapper.deleteHardPatMedOperationItemByOpId(operation.getOpid());
//                count += patMedOperationItemMapper.insertPatMedOperationItems(patMedOperationItems);
//                otherVisitFacade.deleteHardOperationItemByOpId(operation.getOpid());
//                count += otherVisitFacade.insertOperationItems(patMedOperationItems);
//            }catch (Exception ex) {
//                log.info("同步手术详情信息:" + ex.getMessage());
//            }
@@ -491,20 +605,20 @@
                //入院
                pm.setInhospstate("0");
                pm.setCreateTime(new Date());
                patMedInhospMapper.insertPatMedInhosp(pm);
                inhospVisitFacade.insertPatMedInhosp(pm);
            } else if (!Objects.isNull(patMedInhosp.getStartOutHospTime()) && !Objects.isNull(patMedInhosp.getEndOutHospTime())) {
                //出院
                pm.setInhospstate("1");
                pm.setUpdateTime(new Date());
                PatMedInhosp pmi = new PatMedInhosp();
                pmi.setSerialnum(pm.getSerialnum());
                List<PatMedInhosp> patMedInhospList1 = patMedInhospMapper.selectPatMedInhospList(pmi);
                List<PatMedInhosp> patMedInhospList1 = inhospVisitFacade.selectPatMedInhospList(pmi);
                if (CollectionUtils.isEmpty(patMedInhospList1)) {
                    pm.setCreateTime(new Date());
                    patMedInhospMapper.insertPatMedInhosp(pm);
                    inhospVisitFacade.insertPatMedInhosp(pm);
                } else {
                    pm.setInhospid(patMedInhospList1.get(0).getInhospid());
                    patMedInhospMapper.updatePatMedInhosp(pm);
                    inhospVisitFacade.updatePatMedInhosp(pm);
                }

            }