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

[ISSUE #6230] Optimizes ScheduleMessageService code and logic (#6231)

Browse Source 
Co-authored-by: loboxu <loboxu@tencent.com>
This commit is contained in:
Lobo Xu
2023-03-06 10:22:29 +08:00
committed by GitHub
co-authored by loboxu
parent cc8823cf62
commit b9bbd52060
1 changed files with 22 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java
+22 -50
View File
@@ -17,10 +17,7 @@
package org.apache.rocketmq.broker.schedule;
import io.opentelemetry.api.common.Attributes;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
@@ -32,8 +29,8 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
import org.apache.rocketmq.common.ConfigManager;
@@ -96,14 +93,11 @@ public class ScheduleMessageService extends ConfigManager {
this.enableAsyncDeliver = brokerController.getMessageStoreConfig().isEnableScheduleAsyncDeliver();
scheduledPersistService = new ScheduledThreadPoolExecutor(1,
new ThreadFactoryImpl("ScheduleMessageServicePersistThread", true, brokerController.getBrokerConfig()));
scheduledPersistService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
ScheduleMessageService.this.persist();
} catch (Throwable e) {
log.error("scheduleAtFixedRate flush exception", e);
}
scheduledPersistService.scheduleAtFixedRate(() -> {
try {
ScheduleMessageService.this.persist();
} catch (Throwable e) {
log.error("scheduleAtFixedRate flush exception", e);
}
}, 10000, this.brokerController.getMessageStoreConfig().getFlushDelayOffsetInterval(), TimeUnit.MILLISECONDS);
}
@@ -117,9 +111,7 @@ public class ScheduleMessageService extends ConfigManager {
}
public void buildRunningStats(HashMap<String, String> stats) {
Iterator<Map.Entry<Integer, Long>> it = this.offsetTable.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<Integer, Long> next = it.next();
for (Map.Entry<Integer, Long> next : this.offsetTable.entrySet()) {
int queueId = delayLevel2QueueId(next.getKey());
long delayOffset = next.getValue();
long maxOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, queueId);
@@ -169,17 +161,13 @@ public class ScheduleMessageService extends ConfigManager {
}
}
this.deliverExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
if (started.get()) {
ScheduleMessageService.this.persist();
}
} catch (Throwable e) {
log.error("scheduleAtFixedRate flush exception", e);
this.deliverExecutorService.scheduleAtFixedRate(() -> {
try {
if (started.get()) {
ScheduleMessageService.this.persist();
}
} catch (Throwable e) {
log.error("scheduleAtFixedRate flush exception", e);
}
}, 10000, this.brokerController.getMessageStore().getMessageStoreConfig().getFlushDelayOffsetInterval(), TimeUnit.MILLISECONDS);
}
@@ -208,10 +196,8 @@ public class ScheduleMessageService extends ConfigManager {
}
}
if (this.deliverPendingTable != null) {
for (int i = 1; i <= this.deliverPendingTable.size(); i++) {
log.warn("deliverPendingTable level: {}, size: {}", i, this.deliverPendingTable.get(i).size());
}
for (int i = 1; i <= this.deliverPendingTable.size(); i++) {
log.warn("deliverPendingTable level: {}, size: {}", i, this.deliverPendingTable.get(i).size());
}
this.persist();
@@ -378,17 +364,6 @@ public class ScheduleMessageService extends ConfigManager {
return msgInner;
}
public int computeDelayLevel(long timeMillis) {
long intervalMillis = timeMillis - System.currentTimeMillis();
List<Map.Entry<Integer, Long>> sortedLevels = delayLevelTable.entrySet().stream().sorted(Comparator.comparingLong(Map.Entry::getValue)).collect(Collectors.toList());
for (Map.Entry<Integer, Long> entry : sortedLevels) {
if (entry.getValue() > intervalMillis) {
return entry.getKey();
}
}
return sortedLevels.get(sortedLevels.size() - 1).getKey();
}
class DeliverDelayedMessageTimerTask implements Runnable {
private final int delayLevel;
private final long offset;
@@ -402,7 +377,7 @@ public class ScheduleMessageService extends ConfigManager {
public void run() {
try {
if (isStarted()) {
this.executeOnTimeup();
this.executeOnTimeUp();
}
} catch (Exception e) {
// XXX: warn and notify me
@@ -411,9 +386,6 @@ public class ScheduleMessageService extends ConfigManager {
}
}
/**
* @return
*/
private long correctDeliverTimestamp(final long now, final long deliverTimestamp) {
long result = deliverTimestamp;
@@ -426,7 +398,7 @@ public class ScheduleMessageService extends ConfigManager {
return result;
}
public void executeOnTimeup() {
public void executeOnTimeUp() {
ConsumeQueueInterface cq =
ScheduleMessageService.this.brokerController.getMessageStore().getConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
delayLevel2QueueId(delayLevel));
@@ -633,7 +605,7 @@ public class ScheduleMessageService extends ConfigManager {
private boolean autoResend = false;
private CompletableFuture<PutMessageResult> future;
private volatile int resendCount = 0;
private volatile AtomicInteger resendCount = new AtomicInteger(0);
private volatile ProcessStatus status = ProcessStatus.RUNNING;
public PutResultProcess setTopic(String topic) {
@@ -712,7 +684,7 @@ public class ScheduleMessageService extends ConfigManager {
return future;
}
public int getResendCount() {
public AtomicInteger getResendCount() {
return resendCount;
}
@@ -795,7 +767,7 @@ public class ScheduleMessageService extends ConfigManager {
// Gradually increase the resend interval.
try {
Thread.sleep(Math.min(this.resendCount++ * 100, 60 * 1000));
Thread.sleep(Math.min(this.resendCount.incrementAndGet() * 100, 60 * 1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
@@ -823,13 +795,13 @@ public class ScheduleMessageService extends ConfigManager {
public boolean need2Blocked() {
int maxResendNum2Blocked = ScheduleMessageService.this.brokerController.getMessageStore().getMessageStoreConfig()
.getScheduleAsyncDeliverMaxResendNum2Blocked();
return this.resendCount > maxResendNum2Blocked;
return this.resendCount.get() > maxResendNum2Blocked;
}
public boolean need2Skip() {
int maxResendNum2Blocked = ScheduleMessageService.this.brokerController.getMessageStore().getMessageStoreConfig()
.getScheduleAsyncDeliverMaxResendNum2Blocked();
return this.resendCount > maxResendNum2Blocked * 2;
return this.resendCount.get() > maxResendNum2Blocked * 2;
}
@Override