定时消息锁等待

[zonghaishang]

rocketmq5 在往timerwheel中放消息的时候，如果消息需要立即投递到commitlog， 不调用idempotentRelease，不就会导致无用的锁等待么:

1. 直接投递commit log ，不调用 idempotentRelease， 但是后续逻辑会等待锁

        protected void fetchAndPutTimerRequest() throws Exception {
            long tmpCommitQueueOffset = currQueueOffset;
            List<TimerRequest> trs = this.fetchTimerRequests();
            if (CollectionUtils.isEmpty(trs)) {
                commitQueueOffset = tmpCommitQueueOffset;
                maybeMoveWriteTime();
                return;
            }

            while (!isStopped()) {
                CountDownLatch latch = new CountDownLatch(trs.size());
                for (TimerRequest req : trs) {
                    req.setLatch(latch);
                    // @1:  内部调用直接投递commitlog消息
                    this.putMessageToTimerWheel(req);
                }
                // @3:  内部会等待锁，sleep 1秒
                checkDequeueLatch(latch, -1);
                boolean allSuccess = trs.stream().allMatch(TimerRequest::isSucc);
                if (allSuccess) {
                    break;
                } else {
                    holdMomentForUnknownError();
                }
            }
            commitQueueOffset = trs.get(trs.size() - 1).getMsg().getQueueOffset();
            maybeMoveWriteTime();
        }

        protected void putMessageToTimerWheel(TimerRequest req) {
            try {
                perfCounterTicks.startTick(ENQUEUE_PUT);
                DefaultStoreMetricsManager.incTimerEnqueueCount(getRealTopic(req.getMsg()));
                if (shouldRunningDequeue && req.getDelayTime() < currWriteTimeMs) {
                    req.setEnqueueTime(Long.MAX_VALUE);
                    // @2: 这里并没有调用idempotentRelease， 也就是不会递减 CountDownLatch，导致一定卡1秒
                    dequeuePutQueue.put(req);
                } else {
                    boolean doEnqueueRes = doEnqueue(
                        req.getOffsetPy(), req.getSizePy(), req.getDelayTime(), req.getMsg());
                    req.idempotentRelease(doEnqueueRes || storeConfig.isTimerSkipUnknownError());
                }
                perfCounterTicks.endTick(ENQUEUE_PUT);
            } catch (Throwable t) {
                LOGGER.error("Unknown error", t);
                if (storeConfig.isTimerSkipUnknownError()) {
                    req.idempotentRelease(true);
                } else {
                    holdMomentForUnknownError();
                }
            }
        }