[ISSUE #8201] Optimization of Splitting TimeRequest

apache · May 25, 2024 · 0f74efa · 0f74efa
1 parent 97b59a0
commit 0f74efa
Showing 1 changed file with 8 additions and 10 deletions.
diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java
@@ -968,7 +968,6 @@ public int dequeue() throws Exception {
             splitAndCheckLatchForDequeue(deleteMsgStack);
             //read the normal msg
             splitAndCheckLatchForDequeue(normalMsgStack);
-
             // if master -> slave -> master, then the read time move forward, and messages will be lossed
             if (dequeueStatusChangeFlag) {
                 return -1;
@@ -987,7 +986,7 @@ public int dequeue() throws Exception {
     }
 
     private void splitAndCheckLatchForDequeue(List<TimerRequest> origin) throws Exception {
-        if(origin.size() == 0){
+        if (origin.size() == 0) {
             return;
         }
         CountDownLatch latch = new CountDownLatch(origin.size());
@@ -997,26 +996,25 @@ private void splitAndCheckLatchForDequeue(List<TimerRequest> origin) throws Exce
         checkDequeueLatch(latch, currReadTimeMs);
     }
 
-    private void splitIntoQueue(List<TimerRequest> origin,BlockingQueue<List<TimerRequest>> queue, CountDownLatch latch) {
-        if(origin.size() == 0){
-            return;
-        }
+    private void splitIntoQueue(List<TimerRequest> origin, BlockingQueue<List<TimerRequest>> queue,
+        CountDownLatch latch) {
+
         List<TimerRequest> currList = new LinkedList<>();
         int fileIndexPy = -1;
         int msgIndex = 0;
         for (TimerRequest tr : origin) {
             tr.setLatch(latch);
-            if(origin.size() > 100){
-                if(fileIndexPy != tr.getOffsetPy() / commitLogFileSize){
-                    if(currList.size() > 0){
+            if (origin.size() > 100) {
+                if (fileIndexPy != tr.getOffsetPy() / commitLogFileSize) {
+                    if (currList.size() > 0) {
                         queue.add(currList);
                         currList = new LinkedList<>();
                         msgIndex = 0;
                     }
                     fileIndexPy = (int) (tr.getOffsetPy() / commitLogFileSize);
                 }
                 currList.add(tr);
-                if (++msgIndex % 2000 == 0 ) {
+                if (++msgIndex % 2000 == 0) {
                     queue.add(currList);
                     currList = new ArrayList<>();
                 }