update

eight

2024-10-10 340b537e00c8c9a3b2e813d50c470573ab513b5c

 jh-module-ecg/jh-module-ecg-biz/src/main/java/cn/lihu/jh/module/ecg/service/queue/QueueServiceTxFunctions.java

@@ -19,10 +19,7 @@
import org.springframework.validation.annotation.Validated;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@@ -58,8 +55,13 @@
    AtomicInteger openingFlag = new AtomicInteger(0);
    AtomicInteger curSeqNum = new AtomicInteger(0);

    PriorityBlockingQueue<BedQueueBO> priorityQueue = new PriorityBlockingQueue<>();
    ConcurrentHashMap<String, BedQueueBO > mapBedVsQueue = new ConcurrentHashMap<>();

    // 基于类型的, 检查或装机的优先队列
    Map<Integer, PriorityBlockingQueue<BedQueueBO>> mapCheckPriorityQueue = new HashMap();

    // 基于类型的, 装机准备的优先队列
    Map<Integer, PriorityBlockingQueue<BedQueueBO>> mapReadyPriorityQueue = new HashMap();

    Map<Integer, Integer> mapCheckTypeVsReadyMax = null;

@@ -89,7 +91,9 @@
        queueStatusList.add(QueueStatusEnum.READY.getStatus());
        List<QueueDO> queueDOList = queueMapper.getDoctorQueueByStatus(roomId, bedNo, queueStatusList);

        // 新增工位 优先队列
        RoomDO roomDO = roomMapper.getRoom(roomId, bedNo);

        // 新增..工位队列
        BedQueueBO bedQueueBO = new BedQueueBO();
        bedQueueBO.setRoomId(roomId);
        bedQueueBO.setRoomName(roomName);
@@ -97,6 +101,8 @@
        bedQueueBO.setMaxQueueNum(getBedReadyMax(roomId, bedNo));
        bedQueueBO.setQueueNum(new AtomicInteger(queueDOList.size()));
        bedQueueBO.setStatus(BedStatusEnum.OPENING.getStatus());
        bedQueueBO.setCheckTypes( roomDO.getCheckTypes() );
        bedQueueBO.setOpType( roomDO.getOpType() );
        mapBedVsQueue.put(String.format("%09d%s", roomId, bedNo), bedQueueBO);

        return GlobalErrorCodeConstants.SUCCESS;
@@ -133,7 +139,7 @@
        }

        mapBedVsQueue.remove(String.format("%09d%s", roomId, bedNo));
        priorityQueue.remove(bedQueueBO);
        removePriorityQueue(bedQueueBO);
        return GlobalErrorCodeConstants.SUCCESS;
    }

@@ -154,7 +160,7 @@
        }

        bedQueueBO.setStatus(BedStatusEnum.PAUSE.getStatus());
        priorityQueue.remove(bedQueueBO);
        removePriorityQueue(bedQueueBO);
        return GlobalErrorCodeConstants.SUCCESS;
    }

@@ -176,13 +182,7 @@
        }

        bedQueueBO.setStatus(BedStatusEnum.DOCTOR_ON.getStatus());
        if (!priorityQueue.contains(bedQueueBO)) {
            priorityQueue.offer(bedQueueBO);
        } else {
            log.error("bedDoctorResume priorityQueue still exist!. " + roomId + " " + bedNo);
            return ECG_INNER_ERROR;
        }

        addPriorityQueue(bedQueueBO);
        return GlobalErrorCodeConstants.SUCCESS;
    }

@@ -203,13 +203,7 @@
        }

        bedQueueBO.setStatus(BedStatusEnum.DOCTOR_ON.getStatus());
        if (!priorityQueue.contains(bedQueueBO)) {
            priorityQueue.offer(bedQueueBO);
        } else {
            log.error("bedDoctorOn priorityQueue still exist!. " + roomId + " " + bedNo);
            return ECG_INNER_ERROR;
        }

        addPriorityQueue(bedQueueBO);
        return GlobalErrorCodeConstants.SUCCESS;
    }

@@ -230,7 +224,7 @@
            return ROOM_INVALID_STATUS;
        }

        priorityQueue.remove(bedQueueBO);
        removePriorityQueue(bedQueueBO);

        bedQueueBO.setStatus(BedStatusEnum.OPENING.getStatus());
        return GlobalErrorCodeConstants.SUCCESS;
@@ -248,9 +242,8 @@
        // 优先队列中 该工位 就诊准备中人的数量 减一
        BedQueueBO bo = mapBedVsQueue.get(String.format("%09d%s", roomId, bedNo));
        bo.queueNum.getAndDecrement(); // 可能已经【并发的】在 hurry-up 中改变了值
        priorityQueue.remove(bo);
        priorityQueue.offer(bo);

        removePriorityQueue(bo);
        return GlobalErrorCodeConstants.SUCCESS;
    }

@@ -259,7 +252,7 @@
     * 2. 服务运维重启时
     */
    public ErrorCode bedReload() {
        priorityQueue.clear();
        mapCheckPriorityQueue.clear();
        mapBedVsQueue.clear();

        // 清除非当天的排队人员
@@ -290,7 +283,7 @@
            item.queueNum.set( queueNum );
            mapBedVsQueue.put(String.format("%09d%s", item.roomId, item.bedNo), item);
            if (BedStatusEnum.DOCTOR_ON.getStatus() == item.getStatus()) {
                priorityQueue.offer(item);
                addPriorityQueue(item);
            }
        });

@@ -315,39 +308,55 @@
        }
    }

    /**
     * 把医生候诊的队列塞满
     */
    public void hurryup() {
    public void hurryupAllBed() {
        if (0 == openingFlag.get())
            return;

        mapBedVsQueue.values().forEach(bedQueueBO -> hurryupOneBed(bedQueueBO.roomId, bedQueueBO.bedNo));
    }

    /**
     * 把医生候诊的队列塞满
     */
    public void hurryupOneBed(Long roomId, String bedNo) {
        if (0 == openingFlag.get())
            return;

        BedQueueBO bedQueueBO = getBedQueueBO(roomId, bedNo);

        // 处理 过号-回来 的人
        for (BedQueueBO bedQueueBO : mapBedVsQueue.values()) {
            while (bedQueueBO.queueNum.get() < bedQueueBO.maxQueueNum) {
                // 查看 当前工位 是否有过号-回来的患者
                Integer updateNum = queueMapper.queueRecalledPatient(
                        bedQueueBO.getRoomId(),
                        bedQueueBO.getRoomName(),
                        bedQueueBO.getBedNo(),
                        curSeqNum.get() + 1,
                        QueueStatusEnum.RECALLED.getStatus(),
                        QueueStatusEnum.READY.getStatus());
                if (null == updateNum || 0 == updateNum)
                    break;
        while (bedQueueBO.queueNum.get() < bedQueueBO.maxQueueNum) {
            // 查看 当前工位 是否有过号-回来的患者
            Integer updateNum = queueMapper.queueRecalledPatient(
                    bedQueueBO.getRoomId(),
                    bedQueueBO.getRoomName(),
                    bedQueueBO.getBedNo(),
                    curSeqNum.get() + 1,
                    QueueStatusEnum.RECALLED.getStatus(),
                    QueueStatusEnum.READY.getStatus());
            if (null == updateNum || 0 == updateNum)
                break;

                curSeqNum.getAndIncrement();
            curSeqNum.getAndIncrement();

                // 可能已经【并发的】在 nextPatient 中改变了值
                bedQueueBO.queueNum.incrementAndGet();
            // 可能已经【并发的】在 nextPatient 中改变了值
            bedQueueBO.queueNum.incrementAndGet();

                // 可能已经【并发的】在 nextPatient 中改变了优先队列顺序
                priorityQueue.remove(bedQueueBO);
                priorityQueue.offer(bedQueueBO);
            }
            // 可能已经【并发的】在 nextPatient 中改变了优先队列顺序
            refreshPriorityQueue(bedQueueBO);
        }

        // 处理 排队中 患者
        Integer[] checkTypes = bedQueueBO.getCheckTypes();
        Arrays.stream(checkTypes).forEach(checkType -> hurryupOneCheckType(checkType));
    }

    public void hurryupOneCheckType(Integer checkType) {

        PriorityBlockingQueue<BedQueueBO> priorityQueue = mapCheckPriorityQueue.get(checkType);
        if (null == priorityQueue)
            return;

        while (true) {
            BedQueueBO bedQueueBO = priorityQueue.peek();
            if (null == bedQueueBO)
@@ -355,7 +364,7 @@

            int curQueueNum = bedQueueBO.queueNum.get();
            if (curQueueNum > bedQueueBO.maxQueueNum)
                throw new RuntimeException("hurryup: exceed max queue number!");
                throw new RuntimeException("hurryupOneCheckType: exceed max queue number!");

            if (curQueueNum == bedQueueBO.maxQueueNum)
                return;
@@ -367,7 +376,8 @@
                    bedQueueBO.getBedNo(),
                    curSeqNum.get() + 1,
                    QueueStatusEnum.WAITING.getStatus(),
                    QueueStatusEnum.READY.getStatus());
                    QueueStatusEnum.READY.getStatus(),
                    checkType);

            // 没有抢到排队患者
            if (null == updateNum || 0 == updateNum) {
@@ -380,8 +390,7 @@
            bedQueueBO.queueNum.incrementAndGet();

            // 可能已经【并发的】在 nextPatient 中改变了优先队列顺序
            priorityQueue.remove(bedQueueBO);
            priorityQueue.offer(bedQueueBO);
            refreshPriorityQueue(bedQueueBO);
        }
    }

@@ -404,14 +413,15 @@
    public MonitorInfoVO getMonitorInfo() {
        MonitorInfoVO monitorInfoVO = new MonitorInfoVO();
        monitorInfoVO.setOpeningFlag( openingFlag.get() );
        monitorInfoVO.setQueueNum( mapBedVsQueue.size() );
        monitorInfoVO.setActiveQueueNum( priorityQueue.size() );
        roomBedStatistic(monitorInfoVO);

        return monitorInfoVO;
    }

    public void monitorInfo() {
        log.info("map " + mapBedVsQueue.size() + " priority " + priorityQueue.size()
                + " opening " + openingFlag.get());
        MonitorInfoVO monitorInfoVO = new MonitorInfoVO();
        roomBedStatistic(monitorInfoVO);
        log.info(" opening " + openingFlag.get() + " " + monitorInfoVO.getQueueNum() + " " + monitorInfoVO.getActiveQueueNum() + " " + monitorInfoVO.getCheckTypeBedInfo().toString() );
    }

    private Integer getBedReadyMax(Long roomId, String bedNo) {
@@ -424,4 +434,91 @@
        }
        return ret;
    }

    private void addPriorityQueue(BedQueueBO bedQueueBO) {
        Integer[] checkTypes = bedQueueBO.getCheckTypes();
        Arrays.stream(checkTypes).forEach(checkType -> {
            PriorityBlockingQueue<BedQueueBO> priorityQueue = mapCheckPriorityQueue.get(checkType);
            if (!priorityQueue.contains(bedQueueBO)) {
                priorityQueue.offer(bedQueueBO);
            } else {
                log.error("bedDoctorResume priorityQueue still exist!. " + bedQueueBO.getRoomId() + " " + bedQueueBO.getBedNo());
            }
        });
    }

    private void removePriorityQueue(BedQueueBO bedQueueBO) {
        Integer[] checkTypes = bedQueueBO.getCheckTypes();
        Arrays.stream(checkTypes).forEach(checkType -> {
            PriorityBlockingQueue<BedQueueBO> priorityBlockingQueue = mapCheckPriorityQueue.get(checkType);
            if (null != priorityBlockingQueue)
                priorityBlockingQueue.remove(bedQueueBO);
        });
    }

    private void refreshPriorityQueue(BedQueueBO bedQueueBO) {
        Integer[] checkTypes = bedQueueBO.getCheckTypes();
        Arrays.stream(checkTypes).forEach(checkType -> {
            PriorityBlockingQueue<BedQueueBO> priorityQueue = mapCheckPriorityQueue.get(checkType);
            priorityQueue.remove(bedQueueBO);
            priorityQueue.offer(bedQueueBO);
        });
    }

    private void roomBedStatistic(MonitorInfoVO monitorInfoVO) {
        Map<Integer, Integer>  mapOpeningCheckBedStatInfo = new HashMap<>();
        Map<Integer, Integer>  mapOpeningReadyBedStatInfo = new HashMap<>();
        Map<Integer, Integer>  mapOperatingCheckBedStatInfo = new HashMap<>();
        Map<Integer, Integer>  mapOperatingReadyBedStatInfo = new HashMap<>();
        mapCheckTypeVsReadyMax.keySet().forEach( checkType -> {
            mapOpeningCheckBedStatInfo.put(checkType, 0);
            mapOpeningReadyBedStatInfo.put(checkType, 0);
            mapOperatingCheckBedStatInfo.put(checkType, 0);
            mapOperatingReadyBedStatInfo.put(checkType, 0);
        });

        //开通的工位统计
        mapBedVsQueue.values().forEach( bedQueueBO -> {
            Integer[] checkTypes = bedQueueBO.getCheckTypes();
            Arrays.stream(checkTypes).forEach(checkType -> {
                if (bedQueueBO.opType == 0)
                    mapOpeningCheckBedStatInfo.put(checkType, mapOpeningCheckBedStatInfo.get(checkType) + 1 );
                else
                    mapOpeningReadyBedStatInfo.put(checkType, mapOpeningReadyBedStatInfo.get(checkType) + 1 );
            });
        });

        //运营中工位统计
        HashSet<BedQueueBO> hashSetOperatingBedQueue = new HashSet<>();
        mapCheckPriorityQueue.values().forEach(priorityQueue->{
            priorityQueue.stream().forEach(bedQueueBO -> hashSetOperatingBedQueue.add(bedQueueBO));
        });
        mapReadyPriorityQueue.values().forEach(priorityQueue->{
            priorityQueue.stream().forEach(bedQueueBO -> hashSetOperatingBedQueue.add(bedQueueBO));
        });

        hashSetOperatingBedQueue.stream().forEach( bedQueueBO -> {
            Integer[] checkTypes = bedQueueBO.getCheckTypes();
            Arrays.stream(checkTypes).forEach(checkType -> {
                if (bedQueueBO.opType == 0)
                    mapOperatingCheckBedStatInfo.put(checkType, mapOperatingCheckBedStatInfo.get(checkType) + 1 );
                else
                    mapOperatingReadyBedStatInfo.put(checkType, mapOperatingReadyBedStatInfo.get(checkType) + 1 );
            });
        });

        Map<Integer, String> mapCheckTypeBedInfo = new HashMap<>();
        mapCheckTypeVsReadyMax.keySet().forEach( checkType -> {
           String str = mapOpeningCheckBedStatInfo.get(checkType) + " " +
                        mapOpeningReadyBedStatInfo.get(checkType) + " " +
                        mapOperatingCheckBedStatInfo.get(checkType) + " " +
                        mapOperatingReadyBedStatInfo.get(checkType) + " ";
            mapCheckTypeBedInfo.put( checkType, str);
        });

        monitorInfoVO.setCheckTypeBedInfo(mapCheckTypeBedInfo);
        monitorInfoVO.setQueueNum(mapBedVsQueue.size());
        monitorInfoVO.setActiveQueueNum(hashSetOperatingBedQueue.size());
    }

}