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

Refactor MainBatchDispatchRequestService and DispatchService service object locations

Browse Source
This commit is contained in:
nowinkey
2023-02-14 13:50:47 +08:00
committed by Zhouxiang Zhan
parent 87c0a03162
commit 04b8400eb9
1 changed files with 20 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+20 -17
View File
@@ -135,10 +135,6 @@ public class DefaultMessageStore implements MessageStore {
private ReputMessageService reputMessageService;
private MainBatchDispatchRequestService mainBatchDispatchRequestService;
private DispatchService dispatchService;
private HAService haService;
// CompactionLog
@@ -242,8 +238,6 @@ public class DefaultMessageStore implements MessageStore {
this.reputMessageService = new ReputMessageService();
} else {
this.reputMessageService = new ConcurrentReputMessageService();
this.mainBatchDispatchRequestService = new MainBatchDispatchRequestService();
this.dispatchService = new DispatchService();
}
this.transientStorePool = new TransientStorePool(this);
@@ -380,11 +374,6 @@ public class DefaultMessageStore implements MessageStore {
this.reputMessageService.setReputFromOffset(this.commitLog.getConfirmOffset());
this.reputMessageService.start();
if (messageStoreConfig.isEnableBuildConsumeQueueConcurrently()) {
this.mainBatchDispatchRequestService.start();
this.dispatchService.start();
}
// Checking is not necessary, as long as the dLedger's implementation exactly follows the definition of Recover,
// which is eliminating the dispatch inconsistency between the commitLog and consumeQueue at the end of recovery.
this.doRecheckReputOffsetFromCq();
@@ -482,12 +471,7 @@ public class DefaultMessageStore implements MessageStore {
}
this.commitLog.shutdown();
this.reputMessageService.shutdown();
if (mainBatchDispatchRequestService != null) {
mainBatchDispatchRequestService.shutdown();
}
if (dispatchService != null) {
dispatchService.shutdown();
}
this.flushConsumeQueueService.shutdown();
this.allocateMappedFileService.shutdown();
this.storeCheckpoint.flush();
@@ -2983,6 +2967,16 @@ public class DefaultMessageStore implements MessageStore {
private long batchId = 0;
private MainBatchDispatchRequestService mainBatchDispatchRequestService;
private DispatchService dispatchService;
public ConcurrentReputMessageService(){
super();
this.mainBatchDispatchRequestService = new MainBatchDispatchRequestService();
this.dispatchService = new DispatchService();
}
public void createBatchDispatchRequest(ByteBuffer byteBuffer, int position, int size) {
if (position < 0) {
return;
@@ -2992,6 +2986,13 @@ public class DefaultMessageStore implements MessageStore {
batchDispatchRequestQueue.offer(task);
}
@Override
public void start() {
super.start();
this.mainBatchDispatchRequestService.start();
this.dispatchService.start();
}
@Override
public void doReput() {
if (this.reputFromOffset < DefaultMessageStore.this.commitLog.getMinOffset()) {
@@ -3092,6 +3093,8 @@ public class DefaultMessageStore implements MessageStore {
this.reputFromOffset);
}
this.mainBatchDispatchRequestService.shutdown();
this.dispatchService.shutdown();
super.shutdown();
}