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

[ISSUE #6372] new scheduledExecutor clean consume queue (#6376)

Browse Source
This commit is contained in:
fuyou001
2023-03-17 16:48:50 +08:00
committed by GitHub
parent 5ac91199f2
commit 7dbcdc8f96
1 changed files with 19 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+19 -1
View File
@@ -200,6 +200,9 @@ public class DefaultMessageStore implements MessageStore {
private long stateMachineVersion = 0L;
private final ScheduledExecutorService scheduledCleanQueueExecutorService =
Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("StoreCleanQueueScheduledThread"));
public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException {
this.messageArrivingListener = messageArrivingListener;
@@ -456,7 +459,11 @@ public class DefaultMessageStore implements MessageStore {
this.shutdown = true;
this.scheduledExecutorService.shutdown();
this.scheduledCleanQueueExecutorService.shutdown();
try {
this.scheduledExecutorService.awaitTermination(3, TimeUnit.SECONDS);
this.scheduledCleanQueueExecutorService.awaitTermination(3, TimeUnit.SECONDS);
Thread.sleep(1000 * 3);
} catch (InterruptedException e) {
LOGGER.error("shutdown Exception, ", e);
@@ -1771,6 +1778,14 @@ public class DefaultMessageStore implements MessageStore {
}
}, 1, 1, TimeUnit.SECONDS);
this.scheduledCleanQueueExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
DefaultMessageStore.this.cleanQueueFilesPeriodically();
}
}, 1000 * 60, this.messageStoreConfig.getCleanResourceInterval(), TimeUnit.MILLISECONDS);
// this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
// @Override
// public void run() {
@@ -1781,8 +1796,11 @@ public class DefaultMessageStore implements MessageStore {
private void cleanFilesPeriodically() {
this.cleanCommitLogService.run();
this.cleanConsumeQueueService.run();
}
private void cleanQueueFilesPeriodically() {
this.correctLogicOffsetService.run();
this.cleanConsumeQueueService.run();
}
private void checkSelf() {