(rocketmq) branch develop updated: Improve IO for asynchronous delivery processes (#8954)

[lizhanhui]

This is an automated email from the ASF dual-hosted git repository.

lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new c13f051eb8 Improve IO for asynchronous delivery processes (#8954)
c13f051eb8 is described below

commit c13f051eb8deaecc00859028f2e9f398db32b454
Author: hqbfz <125714719+3424672...@users.noreply.github.com>
AuthorDate: Wed Nov 20 11:13:12 2024 +0800

    Improve IO for asynchronous delivery processes (#8954)
    
    Co-authored-by: wanghuaiyuan <wanghuaiy...@xiaomi.com>
---
 store/src/main/java/org/apache/rocketmq/store/CommitLog.java | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java 
b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
index 378518d249..7cf9746551 100644
--- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
@@ -2179,7 +2179,9 @@ public class CommitLog implements Swappable {
             // Asynchronous flush
             else {
                 if 
(!CommitLog.this.defaultMessageStore.isTransientStorePoolEnable()) {
-                    flushCommitLogService.wakeup();
+                    if 
(defaultMessageStore.getMessageStoreConfig().isWakeFlushWhenPutMessage()) {
+                        flushCommitLogService.wakeup();
+                    }
                 } else {
                     if 
(defaultMessageStore.getMessageStoreConfig().isWakeCommitWhenPutMessage()) {
                         commitRealTimeService.wakeup();
@@ -2206,9 +2208,13 @@ public class CommitLog implements Swappable {
             // Asynchronous flush
             else {
                 if 
(!CommitLog.this.defaultMessageStore.isTransientStorePoolEnable()) {
-                    flushCommitLogService.wakeup();
+                    if 
(defaultMessageStore.getMessageStoreConfig().isWakeFlushWhenPutMessage()) {
+                        flushCommitLogService.wakeup();
+                    }
                 } else {
-                    commitRealTimeService.wakeup();
+                    if 
(defaultMessageStore.getMessageStoreConfig().isWakeCommitWhenPutMessage()) {
+                        commitRealTimeService.wakeup();
+                    }
                 }
                 return 
CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
             }