diff --git a/broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java b/broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java
index bdc9c56725..e91c32b55e 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java
@@ -17,10 +17,7 @@
 package org.apache.rocketmq.broker.schedule;
 
 import io.opentelemetry.api.common.Attributes;
-import java.util.Comparator;
 import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
 import java.util.Map;
 import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
@@ -32,8 +29,8 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Collectors;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
 import org.apache.rocketmq.common.ConfigManager;
@@ -96,14 +93,11 @@ public class ScheduleMessageService extends ConfigManager {
         this.enableAsyncDeliver = brokerController.getMessageStoreConfig().isEnableScheduleAsyncDeliver();
         scheduledPersistService = new ScheduledThreadPoolExecutor(1,
             new ThreadFactoryImpl("ScheduleMessageServicePersistThread", true, brokerController.getBrokerConfig()));
-        scheduledPersistService.scheduleAtFixedRate(new Runnable() {
-            @Override
-            public void run() {
-                try {
-                    ScheduleMessageService.this.persist();
-                } catch (Throwable e) {
-                    log.error("scheduleAtFixedRate flush exception", e);
-                }
+        scheduledPersistService.scheduleAtFixedRate(() -> {
+            try {
+                ScheduleMessageService.this.persist();
+            } catch (Throwable e) {
+                log.error("scheduleAtFixedRate flush exception", e);
             }
         }, 10000, this.brokerController.getMessageStoreConfig().getFlushDelayOffsetInterval(), TimeUnit.MILLISECONDS);
     }
@@ -117,9 +111,7 @@ public class ScheduleMessageService extends ConfigManager {
     }
 
     public void buildRunningStats(HashMap<String, String> stats) {
-        Iterator<Map.Entry<Integer, Long>> it = this.offsetTable.entrySet().iterator();
-        while (it.hasNext()) {
-            Map.Entry<Integer, Long> next = it.next();
+        for (Map.Entry<Integer, Long> next : this.offsetTable.entrySet()) {
             int queueId = delayLevel2QueueId(next.getKey());
             long delayOffset = next.getValue();
             long maxOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, queueId);
@@ -169,17 +161,13 @@ public class ScheduleMessageService extends ConfigManager {
                 }
             }
 
-            this.deliverExecutorService.scheduleAtFixedRate(new Runnable() {
-
-                @Override
-                public void run() {
-                    try {
-                        if (started.get()) {
-                            ScheduleMessageService.this.persist();
-                        }
-                    } catch (Throwable e) {
-                        log.error("scheduleAtFixedRate flush exception", e);
+            this.deliverExecutorService.scheduleAtFixedRate(() -> {
+                try {
+                    if (started.get()) {
+                        ScheduleMessageService.this.persist();
                     }
+                } catch (Throwable e) {
+                    log.error("scheduleAtFixedRate flush exception", e);
                 }
             }, 10000, this.brokerController.getMessageStore().getMessageStoreConfig().getFlushDelayOffsetInterval(), TimeUnit.MILLISECONDS);
         }
@@ -208,10 +196,8 @@ public class ScheduleMessageService extends ConfigManager {
                 }
             }
 
-            if (this.deliverPendingTable != null) {
-                for (int i = 1; i <= this.deliverPendingTable.size(); i++) {
-                    log.warn("deliverPendingTable level: {}, size: {}", i, this.deliverPendingTable.get(i).size());
-                }
+            for (int i = 1; i <= this.deliverPendingTable.size(); i++) {
+                log.warn("deliverPendingTable level: {}, size: {}", i, this.deliverPendingTable.get(i).size());
             }
 
             this.persist();
@@ -378,17 +364,6 @@ public class ScheduleMessageService extends ConfigManager {
         return msgInner;
     }
 
-    public int computeDelayLevel(long timeMillis) {
-        long intervalMillis = timeMillis - System.currentTimeMillis();
-        List<Map.Entry<Integer, Long>> sortedLevels = delayLevelTable.entrySet().stream().sorted(Comparator.comparingLong(Map.Entry::getValue)).collect(Collectors.toList());
-        for (Map.Entry<Integer, Long> entry : sortedLevels) {
-            if (entry.getValue() > intervalMillis) {
-                return entry.getKey();
-            }
-        }
-        return sortedLevels.get(sortedLevels.size() - 1).getKey();
-    }
-
     class DeliverDelayedMessageTimerTask implements Runnable {
         private final int delayLevel;
         private final long offset;
@@ -402,7 +377,7 @@ public class ScheduleMessageService extends ConfigManager {
         public void run() {
             try {
                 if (isStarted()) {
-                    this.executeOnTimeup();
+                    this.executeOnTimeUp();
                 }
             } catch (Exception e) {
                 // XXX: warn and notify me
@@ -411,9 +386,6 @@ public class ScheduleMessageService extends ConfigManager {
             }
         }
 
-        /**
-         * @return
-         */
         private long correctDeliverTimestamp(final long now, final long deliverTimestamp) {
 
             long result = deliverTimestamp;
@@ -426,7 +398,7 @@ public class ScheduleMessageService extends ConfigManager {
             return result;
         }
 
-        public void executeOnTimeup() {
+        public void executeOnTimeUp() {
             ConsumeQueueInterface cq =
                 ScheduleMessageService.this.brokerController.getMessageStore().getConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
                     delayLevel2QueueId(delayLevel));
@@ -633,7 +605,7 @@ public class ScheduleMessageService extends ConfigManager {
         private boolean autoResend = false;
         private CompletableFuture<PutMessageResult> future;
 
-        private volatile int resendCount = 0;
+        private volatile AtomicInteger resendCount = new AtomicInteger(0);
         private volatile ProcessStatus status = ProcessStatus.RUNNING;
 
         public PutResultProcess setTopic(String topic) {
@@ -712,7 +684,7 @@ public class ScheduleMessageService extends ConfigManager {
             return future;
         }
 
-        public int getResendCount() {
+        public AtomicInteger getResendCount() {
             return resendCount;
         }
 
@@ -795,7 +767,7 @@ public class ScheduleMessageService extends ConfigManager {
 
             // Gradually increase the resend interval.
             try {
-                Thread.sleep(Math.min(this.resendCount++ * 100, 60 * 1000));
+                Thread.sleep(Math.min(this.resendCount.incrementAndGet() * 100, 60 * 1000));
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }
@@ -823,13 +795,13 @@ public class ScheduleMessageService extends ConfigManager {
         public boolean need2Blocked() {
             int maxResendNum2Blocked = ScheduleMessageService.this.brokerController.getMessageStore().getMessageStoreConfig()
                 .getScheduleAsyncDeliverMaxResendNum2Blocked();
-            return this.resendCount > maxResendNum2Blocked;
+            return this.resendCount.get() > maxResendNum2Blocked;
         }
 
         public boolean need2Skip() {
             int maxResendNum2Blocked = ScheduleMessageService.this.brokerController.getMessageStore().getMessageStoreConfig()
                 .getScheduleAsyncDeliverMaxResendNum2Blocked();
-            return this.resendCount > maxResendNum2Blocked * 2;
+            return this.resendCount.get() > maxResendNum2Blocked * 2;
         }
 
         @Override