opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-28 20:09:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #7201] Remove the DefaultMessageStore.class dependency in TransientStorePool

Browse Source 
Co-authored-by: guyinyou <guyinyou.gyy@alibaba-inc.com>
This commit is contained in:
guyinyou
2023-08-17 13:59:04 +08:00
committed by GitHub
parent a4bcc2a74d
commit 3df1b9232a
3 changed files with 12 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/AllocateMappedFileService.java
+3 -3
View File
@@ -55,7 +55,7 @@ public class AllocateMappedFileService extends ServiceThread {
if (this.messageStore.isTransientStorePoolEnable()) {
if (this.messageStore.getMessageStoreConfig().isFastFailIfNoBufferInStorePool()
&& BrokerRole.SLAVE != this.messageStore.getMessageStoreConfig().getBrokerRole()) { //if broker is slave, don't fast fail even no buffer in pool
canSubmitRequests = this.messageStore.getTransientStorePool().availableBufferNums() - this.requestQueue.size();
canSubmitRequests = this.messageStore.remainTransientStoreBufferNumbs() - this.requestQueue.size();
}
}
@@ -65,7 +65,7 @@ public class AllocateMappedFileService extends ServiceThread {
if (nextPutOK) {
if (canSubmitRequests <= 0) {
log.warn("[NOTIFYME]TransientStorePool is not enough, so create mapped file error, " +
"RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.getTransientStorePool().availableBufferNums());
"RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.remainTransientStoreBufferNumbs());
this.requestTable.remove(nextFilePath);
return null;
}
@@ -81,7 +81,7 @@ public class AllocateMappedFileService extends ServiceThread {
if (nextNextPutOK) {
if (canSubmitRequests <= 0) {
log.warn("[NOTIFYME]TransientStorePool is not enough, so skip preallocate mapped file, " +
"RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.getTransientStorePool().availableBufferNums());
"RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.remainTransientStoreBufferNumbs());
this.requestTable.remove(nextNextFilePath);
} else {
boolean offerOK = this.requestQueue.offer(nextNextReq);
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+5 -2
View File
@@ -250,7 +250,7 @@ public class DefaultMessageStore implements MessageStore {
this.reputMessageService = new ConcurrentReputMessageService();
}
this.transientStorePool = new TransientStorePool(this);
this.transientStorePool = new TransientStorePool(messageStoreConfig.getTransientStorePoolSize(), messageStoreConfig.getMappedFileSizeCommitLog());
this.scheduledExecutorService =
Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("StoreScheduledThread", getBrokerIdentity()));
@@ -1983,7 +1983,10 @@ public class DefaultMessageStore implements MessageStore {
}
public int remainTransientStoreBufferNumbs() {
return this.transientStorePool.availableBufferNums();
if (this.isTransientStorePoolEnable()) {
return this.transientStorePool.availableBufferNums();
}
return Integer.MAX_VALUE;
}
@Override
store/src/main/java/org/apache/rocketmq/store/TransientStorePool.java
+4 -9
View File
@@ -33,13 +33,11 @@ public class TransientStorePool {
private final int poolSize;
private final int fileSize;
private final Deque<ByteBuffer> availableBuffers;
private final DefaultMessageStore messageStore;
private volatile boolean isRealCommit = true;
public TransientStorePool(final DefaultMessageStore messageStore) {
this.messageStore = messageStore;
this.poolSize = messageStore.getMessageStoreConfig().getTransientStorePoolSize();
this.fileSize = messageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
public TransientStorePool(final int poolSize, final int fileSize) {
this.poolSize = poolSize;
this.fileSize = fileSize;
this.availableBuffers = new ConcurrentLinkedDeque<>();
}
@@ -81,10 +79,7 @@ public class TransientStorePool {
}
public int availableBufferNums() {
if (messageStore.isTransientStorePoolEnable()) {
return availableBuffers.size();
}
return Integer.MAX_VALUE;
return availableBuffers.size();
}
public boolean isRealCommit() {