diff --git a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
index 1188f2191f..bb1e01f247 100644
--- a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
+++ b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
@@ -163,6 +163,10 @@ public class MessageStoreConfig {
private boolean enableMultiDispatch = false;
private int maxLmqConsumeQueueNum = 20000;
+ private boolean enableScheduleAsyncDeliver = false;
+ private int scheduleAsyncDeliverMaxPendingLimit = 2000;
+ private int scheduleAsyncDeliverMaxResendNum2Blocked = 3;
+
public boolean isDebugLockEnable() {
return debugLockEnable;
}
@@ -772,4 +776,28 @@ public class MessageStoreConfig {
public void setMaxLmqConsumeQueueNum(int maxLmqConsumeQueueNum) {
this.maxLmqConsumeQueueNum = maxLmqConsumeQueueNum;
}
+
+ public boolean isEnableScheduleAsyncDeliver() {
+ return enableScheduleAsyncDeliver;
+ }
+
+ public void setEnableScheduleAsyncDeliver(boolean enableScheduleAsyncDeliver) {
+ this.enableScheduleAsyncDeliver = enableScheduleAsyncDeliver;
+ }
+
+ public int getScheduleAsyncDeliverMaxPendingLimit() {
+ return scheduleAsyncDeliverMaxPendingLimit;
+ }
+
+ public void setScheduleAsyncDeliverMaxPendingLimit(int scheduleAsyncDeliverMaxPendingLimit) {
+ this.scheduleAsyncDeliverMaxPendingLimit = scheduleAsyncDeliverMaxPendingLimit;
+ }
+
+ public int getScheduleAsyncDeliverMaxResendNum2Blocked() {
+ return scheduleAsyncDeliverMaxResendNum2Blocked;
+ }
+
+ public void setScheduleAsyncDeliverMaxResendNum2Blocked(int scheduleAsyncDeliverMaxResendNum2Blocked) {
+ this.scheduleAsyncDeliverMaxResendNum2Blocked = scheduleAsyncDeliverMaxResendNum2Blocked;
+ }
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java b/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java
index dd68c73c37..c08c515c0c 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java
@@ -168,7 +168,7 @@ public class HAConnection {
if (HAConnection.this.slaveRequestOffset < 0) {
HAConnection.this.slaveRequestOffset = readOffset;
log.info("slave[" + HAConnection.this.clientAddr + "] request offset " + readOffset);
- } else if (HAConnection.this.slaveAckOffset > HAConnection.this.haService.getDefaultMessageStore().getMaxPhyOffset()){
+ } else if (HAConnection.this.slaveAckOffset > HAConnection.this.haService.getDefaultMessageStore().getMaxPhyOffset()) {
log.warn("slave[{}] request offset={} greater than local commitLog offset={}. ",
HAConnection.this.clientAddr,
HAConnection.this.slaveAckOffset,
diff --git a/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java b/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
index c45287ff65..d5b4e8d390 100644
--- a/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
+++ b/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
@@ -19,8 +19,12 @@ package org.apache.rocketmq.store.schedule;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -54,6 +58,8 @@ public class ScheduleMessageService extends ConfigManager {
private static final long FIRST_DELAY_TIME = 1000L;
private static final long DELAY_FOR_A_WHILE = 100L;
private static final long DELAY_FOR_A_PERIOD = 10000L;
+ private static final long WAIT_FOR_SHUTDOWN = 5000L;
+ private static final long DELAY_FOR_A_SLEEP = 10L;
private final ConcurrentMap delayLevelTable =
new ConcurrentHashMap(32);
@@ -63,13 +69,19 @@ public class ScheduleMessageService extends ConfigManager {
private final DefaultMessageStore defaultMessageStore;
private final AtomicBoolean started = new AtomicBoolean(false);
private ScheduledExecutorService deliverExecutorService;
- private int deliverThreadPoolNums = Runtime.getRuntime().availableProcessors();
private MessageStore writeMessageStore;
private int maxDelayLevel;
+ private boolean enableAsyncDeliver = false;
+ private ScheduledExecutorService handleExecutorService;
+ private final Map> deliverPendingTable =
+ new ConcurrentHashMap<>(32);
public ScheduleMessageService(final DefaultMessageStore defaultMessageStore) {
this.defaultMessageStore = defaultMessageStore;
this.writeMessageStore = defaultMessageStore;
+ if (defaultMessageStore != null) {
+ this.enableAsyncDeliver = defaultMessageStore.getMessageStoreConfig().isEnableScheduleAsyncDeliver();
+ }
}
public static int queueId2DelayLevel(final int queueId) {
@@ -116,7 +128,10 @@ public class ScheduleMessageService extends ConfigManager {
public void start() {
if (started.compareAndSet(false, true)) {
super.load();
- this.deliverExecutorService = new ScheduledThreadPoolExecutor(deliverThreadPoolNums, new ThreadFactoryImpl("ScheduleMessageTimerThread_"));
+ this.deliverExecutorService = new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl("ScheduleMessageTimerThread_"));
+ if (this.enableAsyncDeliver) {
+ this.handleExecutorService = new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl("ScheduleMessageExecutorHandleThread_"));
+ }
for (Map.Entry entry : this.delayLevelTable.entrySet()) {
Integer level = entry.getKey();
Long timeDelay = entry.getValue();
@@ -126,6 +141,9 @@ public class ScheduleMessageService extends ConfigManager {
}
if (timeDelay != null) {
+ if (this.enableAsyncDeliver) {
+ this.handleExecutorService.schedule(new HandlePutResultTask(level), FIRST_DELAY_TIME, TimeUnit.MILLISECONDS);
+ }
this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME, TimeUnit.MILLISECONDS);
}
}
@@ -148,7 +166,29 @@ public class ScheduleMessageService extends ConfigManager {
public void shutdown() {
if (this.started.compareAndSet(true, false) && null != this.deliverExecutorService) {
- this.deliverExecutorService.shutdownNow();
+ this.deliverExecutorService.shutdown();
+ try {
+ this.deliverExecutorService.awaitTermination(WAIT_FOR_SHUTDOWN, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ log.error("deliverExecutorService awaitTermination error", e);
+ }
+
+ if (this.handleExecutorService != null) {
+ this.handleExecutorService.shutdown();
+ try {
+ this.handleExecutorService.awaitTermination(WAIT_FOR_SHUTDOWN, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ log.error("handleExecutorService awaitTermination error", e);
+ }
+ }
+
+ if (this.deliverPendingTable != null) {
+ for (int i = 1; i <= this.deliverPendingTable.size(); i++) {
+ log.warn("deliverPendingTable level: {}, size: {}", i, this.deliverPendingTable.get(i).size());
+ }
+ }
+
+ this.persist();
}
}
@@ -255,6 +295,9 @@ public class ScheduleMessageService extends ConfigManager {
long num = Long.parseLong(value.substring(0, value.length() - 1));
long delayTimeMillis = tu * num;
this.delayLevelTable.put(level, delayTimeMillis);
+ if (this.enableAsyncDeliver) {
+ this.deliverPendingTable.put(level, new LinkedBlockingQueue<>());
+ }
}
} catch (Exception e) {
log.error("parseDelayLevel exception", e);
@@ -265,6 +308,36 @@ public class ScheduleMessageService extends ConfigManager {
return true;
}
+ private MessageExtBrokerInner messageTimeup(MessageExt msgExt) {
+ MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
+ msgInner.setBody(msgExt.getBody());
+ msgInner.setFlag(msgExt.getFlag());
+ MessageAccessor.setProperties(msgInner, msgExt.getProperties());
+
+ TopicFilterType topicFilterType = MessageExt.parseTopicFilterType(msgInner.getSysFlag());
+ long tagsCodeValue =
+ MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
+ msgInner.setTagsCode(tagsCodeValue);
+ msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
+
+ msgInner.setSysFlag(msgExt.getSysFlag());
+ msgInner.setBornTimestamp(msgExt.getBornTimestamp());
+ msgInner.setBornHost(msgExt.getBornHost());
+ msgInner.setStoreHost(msgExt.getStoreHost());
+ msgInner.setReconsumeTimes(msgExt.getReconsumeTimes());
+
+ msgInner.setWaitStoreMsgOK(false);
+ MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
+
+ msgInner.setTopic(msgInner.getProperty(MessageConst.PROPERTY_REAL_TOPIC));
+
+ String queueIdStr = msgInner.getProperty(MessageConst.PROPERTY_REAL_QUEUE_ID);
+ int queueId = Integer.parseInt(queueIdStr);
+ msgInner.setQueueId(queueId);
+
+ return msgInner;
+ }
+
class DeliverDelayedMessageTimerTask implements Runnable {
private final int delayLevel;
private final long offset;
@@ -283,8 +356,7 @@ public class ScheduleMessageService extends ConfigManager {
} catch (Exception e) {
// XXX: warn and notify me
log.error("ScheduleMessageService, executeOnTimeup exception", e);
- ScheduleMessageService.this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(
- this.delayLevel, this.offset), DELAY_FOR_A_PERIOD, TimeUnit.MILLISECONDS);
+ this.scheduleNextTimerTask(this.offset, DELAY_FOR_A_PERIOD);
}
}
@@ -308,158 +380,431 @@ public class ScheduleMessageService extends ConfigManager {
ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
delayLevel2QueueId(delayLevel));
- long failScheduleOffset = offset;
+ if (cq == null) {
+ this.scheduleNextTimerTask(this.offset, DELAY_FOR_A_WHILE);
+ return;
+ }
- if (cq != null) {
- SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
- if (bufferCQ != null) {
- try {
- long nextOffset = offset;
- int i = 0;
- ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
- for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
- long offsetPy = bufferCQ.getByteBuffer().getLong();
- int sizePy = bufferCQ.getByteBuffer().getInt();
- long tagsCode = bufferCQ.getByteBuffer().getLong();
+ SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
+ if (bufferCQ == null) {
+ long resetOffset;
+ if ((resetOffset = cq.getMinOffsetInQueue()) > this.offset) {
+ log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, queueId={}",
+ this.offset, resetOffset, cq.getQueueId());
+ } else if ((resetOffset = cq.getMaxOffsetInQueue()) < this.offset) {
+ log.error("schedule CQ offset invalid. offset={}, cqMaxOffset={}, queueId={}",
+ this.offset, resetOffset, cq.getQueueId());
+ } else {
+ resetOffset = this.offset;
+ }
- if (cq.isExtAddr(tagsCode)) {
- if (cq.getExt(tagsCode, cqExtUnit)) {
- tagsCode = cqExtUnit.getTagsCode();
- } else {
- //can't find ext content.So re compute tags code.
- log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}",
- tagsCode, offsetPy, sizePy);
- long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy);
- tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime);
- }
- }
+ this.scheduleNextTimerTask(resetOffset, DELAY_FOR_A_WHILE);
+ return;
+ }
- long now = System.currentTimeMillis();
- long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
+ long nextOffset = this.offset;
+ try {
+ int i = 0;
+ ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
+ for (; i < bufferCQ.getSize() && isStarted(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
+ long offsetPy = bufferCQ.getByteBuffer().getLong();
+ int sizePy = bufferCQ.getByteBuffer().getInt();
+ long tagsCode = bufferCQ.getByteBuffer().getLong();
- nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
+ if (cq.isExtAddr(tagsCode)) {
+ if (cq.getExt(tagsCode, cqExtUnit)) {
+ tagsCode = cqExtUnit.getTagsCode();
+ } else {
+ //can't find ext content.So re compute tags code.
+ log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}",
+ tagsCode, offsetPy, sizePy);
+ long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy);
+ tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime);
+ }
+ }
- long countdown = deliverTimestamp - now;
+ long now = System.currentTimeMillis();
+ long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
+ nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
- if (countdown <= 0) {
- MessageExt msgExt =
- ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(
- offsetPy, sizePy);
-
- if (msgExt != null) {
- try {
- MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
- if (TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC.equals(msgInner.getTopic())) {
- log.error("[BUG] the real topic of schedule msg is {}, discard the msg. msg={}",
- msgInner.getTopic(), msgInner);
- continue;
- }
- PutMessageResult putMessageResult =
- ScheduleMessageService.this.writeMessageStore
- .putMessage(msgInner);
-
- if (putMessageResult != null
- && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
- if (ScheduleMessageService.this.defaultMessageStore.getMessageStoreConfig().isEnableScheduleMessageStats()) {
- ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incQueueGetNums(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, delayLevel - 1, putMessageResult.getAppendMessageResult().getMsgNum());
- ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incQueueGetSize(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, delayLevel - 1, putMessageResult.getAppendMessageResult().getWroteBytes());
- ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incGroupGetNums(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, putMessageResult.getAppendMessageResult().getMsgNum());
- ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incGroupGetSize(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, putMessageResult.getAppendMessageResult().getWroteBytes());
- ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incTopicPutNums(msgInner.getTopic(), putMessageResult.getAppendMessageResult().getMsgNum(), 1);
- ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incTopicPutSize(msgInner.getTopic(),
- putMessageResult.getAppendMessageResult().getWroteBytes());
- ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incBrokerPutNums(putMessageResult.getAppendMessageResult().getMsgNum());
- }
- continue;
- } else {
- // XXX: warn and notify me
- log.error(
- "ScheduleMessageService, a message time up, but reput it failed, topic: {} msgId {}",
- msgExt.getTopic(), msgExt.getMsgId());
- ScheduleMessageService.this.deliverExecutorService.schedule(
- new DeliverDelayedMessageTimerTask(this.delayLevel,
- nextOffset), DELAY_FOR_A_PERIOD, TimeUnit.MILLISECONDS);
- ScheduleMessageService.this.updateOffset(this.delayLevel,
- nextOffset);
- return;
- }
- } catch (Exception e) {
- /*
- * XXX: warn and notify me
- */
- log.error(
- "ScheduleMessageService, messageTimeup execute error, drop it. msgExt={}, nextOffset={}, offsetPy={}, sizePy={}", msgExt, nextOffset, offsetPy, sizePy, e);
- }
- }
- } else {
- ScheduleMessageService.this.deliverExecutorService.schedule(
- new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset),
- countdown, TimeUnit.MILLISECONDS);
- ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
- return;
- }
- } // end of for
-
- nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
- ScheduleMessageService.this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(
- this.delayLevel, nextOffset), DELAY_FOR_A_WHILE, TimeUnit.MILLISECONDS);
- ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
+ long countdown = deliverTimestamp - now;
+ if (countdown > 0) {
+ this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
return;
- } finally {
-
- bufferCQ.release();
- }
- } // end of if (bufferCQ != null)
- else {
-
- long cqMinOffset = cq.getMinOffsetInQueue();
- long cqMaxOffset = cq.getMaxOffsetInQueue();
- if (offset < cqMinOffset) {
- failScheduleOffset = cqMinOffset;
- log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, cqMaxOffset={}, queueId={}",
- offset, cqMinOffset, cqMaxOffset, cq.getQueueId());
}
- if (offset > cqMaxOffset) {
- failScheduleOffset = cqMaxOffset;
- log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, cqMaxOffset={}, queueId={}",
- offset, cqMinOffset, cqMaxOffset, cq.getQueueId());
+ MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(offsetPy, sizePy);
+ if (msgExt == null) {
+ continue;
+ }
+
+ MessageExtBrokerInner msgInner = ScheduleMessageService.this.messageTimeup(msgExt);
+ if (TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC.equals(msgInner.getTopic())) {
+ log.error("[BUG] the real topic of schedule msg is {}, discard the msg. msg={}",
+ msgInner.getTopic(), msgInner);
+ continue;
+ }
+
+ boolean deliverSuc;
+ if (ScheduleMessageService.this.enableAsyncDeliver) {
+ deliverSuc = this.asyncDeliver(msgInner, msgExt.getMsgId(), offset, offsetPy, sizePy);
+ } else {
+ deliverSuc = this.syncDeliver(msgInner, msgExt.getMsgId(), offset, offsetPy, sizePy);
+ }
+
+ if (!deliverSuc) {
+ this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
+ return;
}
}
- } // end of if (cq != null)
- ScheduleMessageService.this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel,
- failScheduleOffset), DELAY_FOR_A_WHILE, TimeUnit.MILLISECONDS);
+ nextOffset = this.offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
+ } catch (Exception e) {
+ log.error("ScheduleMessageService, messageTimeup execute error, offset = {}", nextOffset, e);
+ } finally {
+ bufferCQ.release();
+ }
+
+ this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
}
- private MessageExtBrokerInner messageTimeup(MessageExt msgExt) {
- MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
- msgInner.setBody(msgExt.getBody());
- msgInner.setFlag(msgExt.getFlag());
- MessageAccessor.setProperties(msgInner, msgExt.getProperties());
+ public void scheduleNextTimerTask(long offset, long delay) {
+ ScheduleMessageService.this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(
+ this.delayLevel, offset), delay, TimeUnit.MILLISECONDS);
+ }
- TopicFilterType topicFilterType = MessageExt.parseTopicFilterType(msgInner.getSysFlag());
- long tagsCodeValue =
- MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
- msgInner.setTagsCode(tagsCodeValue);
- msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
+ private boolean syncDeliver(MessageExtBrokerInner msgInner, String msgId, long offset, long offsetPy,
+ int sizePy) {
+ PutResultProcess resultProcess = deliverMessage(msgInner, msgId, offset, offsetPy, sizePy, false);
+ PutMessageResult result = resultProcess.get();
+ boolean sendStatus = result != null && result.getPutMessageStatus() == PutMessageStatus.PUT_OK;
+ if (sendStatus) {
+ ScheduleMessageService.this.updateOffset(this.delayLevel, resultProcess.getNextOffset());
+ }
+ return sendStatus;
+ }
- msgInner.setSysFlag(msgExt.getSysFlag());
- msgInner.setBornTimestamp(msgExt.getBornTimestamp());
- msgInner.setBornHost(msgExt.getBornHost());
- msgInner.setStoreHost(msgExt.getStoreHost());
- msgInner.setReconsumeTimes(msgExt.getReconsumeTimes());
+ private boolean asyncDeliver(MessageExtBrokerInner msgInner, String msgId, long offset, long offsetPy,
+ int sizePy) {
+ Queue processesQueue = ScheduleMessageService.this.deliverPendingTable.get(this.delayLevel);
- msgInner.setWaitStoreMsgOK(false);
- MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
+ //Flow Control
+ int currentPendingNum = processesQueue.size();
+ int maxPendingLimit = ScheduleMessageService.this.defaultMessageStore.getMessageStoreConfig()
+ .getScheduleAsyncDeliverMaxPendingLimit();
+ if (currentPendingNum > maxPendingLimit) {
+ log.warn("Asynchronous deliver triggers flow control, " +
+ "currentPendingNum={}, maxPendingLimit={}", currentPendingNum, maxPendingLimit);
+ return false;
+ }
- msgInner.setTopic(msgInner.getProperty(MessageConst.PROPERTY_REAL_TOPIC));
+ //Blocked
+ PutResultProcess firstProcess = processesQueue.peek();
+ if (firstProcess != null && firstProcess.need2Blocked()) {
+ log.warn("Asynchronous deliver block. info={}", firstProcess.toString());
+ return false;
+ }
- String queueIdStr = msgInner.getProperty(MessageConst.PROPERTY_REAL_QUEUE_ID);
- int queueId = Integer.parseInt(queueIdStr);
- msgInner.setQueueId(queueId);
+ PutResultProcess resultProcess = deliverMessage(msgInner, msgId, offset, offsetPy, sizePy, true);
+ processesQueue.add(resultProcess);
+ return true;
+ }
- return msgInner;
+ private PutResultProcess deliverMessage(MessageExtBrokerInner msgInner, String msgId, long offset,
+ long offsetPy, int sizePy, boolean autoResend) {
+ CompletableFuture future =
+ ScheduleMessageService.this.writeMessageStore.asyncPutMessage(msgInner);
+ return new PutResultProcess()
+ .setTopic(msgInner.getTopic())
+ .setDelayLevel(this.delayLevel)
+ .setOffset(offset)
+ .setPhysicOffset(offsetPy)
+ .setPhysicSize(sizePy)
+ .setMsgId(msgId)
+ .setAutoResend(autoResend)
+ .setFuture(future)
+ .thenProcess();
}
}
+
+ public class HandlePutResultTask implements Runnable {
+ private final int delayLevel;
+
+ public HandlePutResultTask(int delayLevel) {
+ this.delayLevel = delayLevel;
+ }
+
+ @Override
+ public void run() {
+ LinkedBlockingQueue pendingQueue =
+ ScheduleMessageService.this.deliverPendingTable.get(this.delayLevel);
+
+ PutResultProcess putResultProcess;
+ while ((putResultProcess = pendingQueue.peek()) != null) {
+ try {
+ switch (putResultProcess.getStatus()) {
+ case SUCCESS:
+ ScheduleMessageService.this.updateOffset(this.delayLevel, putResultProcess.getNextOffset());
+ pendingQueue.remove();
+ break;
+ case RUNNING:
+ break;
+ case EXCEPTION:
+ if (!isStarted()) {
+ log.warn("HandlePutResultTask shutdown, info={}", putResultProcess.toString());
+ return;
+ }
+ log.warn("putResultProcess error, info={}", putResultProcess.toString());
+ putResultProcess.onException();
+ break;
+ case SKIP:
+ log.warn("putResultProcess skip, info={}", putResultProcess.toString());
+ pendingQueue.remove();
+ break;
+ }
+ } catch (Exception e) {
+ log.error("HandlePutResultTask exception. info={}", putResultProcess.toString(), e);
+ putResultProcess.onException();
+ }
+ }
+
+ if (isStarted()) {
+ ScheduleMessageService.this.handleExecutorService
+ .schedule(new HandlePutResultTask(this.delayLevel), DELAY_FOR_A_SLEEP, TimeUnit.MILLISECONDS);
+ }
+ }
+ }
+
+ public class PutResultProcess {
+ private String topic;
+ private long offset;
+ private long physicOffset;
+ private int physicSize;
+ private int delayLevel;
+ private String msgId;
+ private boolean autoResend = false;
+ private CompletableFuture future;
+
+ private volatile int resendCount = 0;
+ private volatile ProcessStatus status = ProcessStatus.RUNNING;
+
+ public PutResultProcess setTopic(String topic) {
+ this.topic = topic;
+ return this;
+ }
+
+ public PutResultProcess setOffset(long offset) {
+ this.offset = offset;
+ return this;
+ }
+
+ public PutResultProcess setPhysicOffset(long physicOffset) {
+ this.physicOffset = physicOffset;
+ return this;
+ }
+
+ public PutResultProcess setPhysicSize(int physicSize) {
+ this.physicSize = physicSize;
+ return this;
+ }
+
+ public PutResultProcess setDelayLevel(int delayLevel) {
+ this.delayLevel = delayLevel;
+ return this;
+ }
+
+ public PutResultProcess setMsgId(String msgId) {
+ this.msgId = msgId;
+ return this;
+ }
+
+ public PutResultProcess setAutoResend(boolean autoResend) {
+ this.autoResend = autoResend;
+ return this;
+ }
+
+ public PutResultProcess setFuture(CompletableFuture future) {
+ this.future = future;
+ return this;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public long getOffset() {
+ return offset;
+ }
+
+ public long getNextOffset() {
+ return offset + 1;
+ }
+
+ public long getPhysicOffset() {
+ return physicOffset;
+ }
+
+ public int getPhysicSize() {
+ return physicSize;
+ }
+
+ public Integer getDelayLevel() {
+ return delayLevel;
+ }
+
+ public String getMsgId() {
+ return msgId;
+ }
+
+ public boolean isAutoResend() {
+ return autoResend;
+ }
+
+ public CompletableFuture getFuture() {
+ return future;
+ }
+
+ public int getResendCount() {
+ return resendCount;
+ }
+
+ public PutResultProcess thenProcess() {
+ this.future.thenAccept(result -> {
+ this.handleResult(result);
+ });
+
+ this.future.exceptionally(e -> {
+ log.error("ScheduleMessageService put message exceptionally, info: {}",
+ PutResultProcess.this.toString(), e);
+
+ onException();
+ return null;
+ });
+ return this;
+ }
+
+ private void handleResult(PutMessageResult result) {
+ if (result != null && result.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
+ onSuccess(result);
+ } else {
+ log.warn("ScheduleMessageService put message failed. info: {}.", result);
+ onException();
+ }
+ }
+
+ public void onSuccess(PutMessageResult result) {
+ this.status = ProcessStatus.SUCCESS;
+ if (ScheduleMessageService.this.defaultMessageStore.getMessageStoreConfig().isEnableScheduleMessageStats()) {
+ ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incQueueGetNums(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, delayLevel - 1, result.getAppendMessageResult().getMsgNum());
+ ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incQueueGetSize(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, delayLevel - 1, result.getAppendMessageResult().getWroteBytes());
+ ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incGroupGetNums(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, result.getAppendMessageResult().getMsgNum());
+ ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incGroupGetSize(MixAll.SCHEDULE_CONSUMER_GROUP, TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, result.getAppendMessageResult().getWroteBytes());
+ ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incTopicPutNums(this.topic, result.getAppendMessageResult().getMsgNum(), 1);
+ ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incTopicPutSize(this.topic, result.getAppendMessageResult().getWroteBytes());
+ ScheduleMessageService.this.defaultMessageStore.getBrokerStatsManager().incBrokerPutNums(result.getAppendMessageResult().getMsgNum());
+ }
+ }
+
+ public void onException() {
+ log.warn("ScheduleMessageService onException, info: {}", this.toString());
+ if (this.autoResend) {
+ this.resend();
+ } else {
+ this.status = ProcessStatus.SKIP;
+ }
+ }
+
+ public ProcessStatus getStatus() {
+ return this.status;
+ }
+
+ public PutMessageResult get() {
+ try {
+ return this.future.get();
+ } catch (InterruptedException | ExecutionException e) {
+ return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, null);
+ }
+ }
+
+ private void resend() {
+ log.info("Resend message, info: {}", this.toString());
+
+ // Gradually increase the resend interval.
+ try {
+ Thread.sleep(Math.min(this.resendCount++ * 100, 60 * 1000));
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ try {
+ MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(this.physicOffset, this.physicSize);
+ if (msgExt == null) {
+ log.warn("ScheduleMessageService resend not found message. info: {}", this.toString());
+ this.status = need2Skip() ? ProcessStatus.SKIP : ProcessStatus.EXCEPTION;
+ return;
+ }
+
+ MessageExtBrokerInner msgInner = ScheduleMessageService.this.messageTimeup(msgExt);
+ PutMessageResult result = ScheduleMessageService.this.writeMessageStore.putMessage(msgInner);
+ this.handleResult(result);
+ if (result != null && result.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
+ log.info("Resend message success, info: {}", this.toString());
+ }
+ } catch (Exception e) {
+ this.status = ProcessStatus.EXCEPTION;
+ log.error("Resend message error, info: {}", this.toString(), e);
+ }
+ }
+
+ public boolean need2Blocked() {
+ int maxResendNum2Blocked = ScheduleMessageService.this.defaultMessageStore.getMessageStoreConfig()
+ .getScheduleAsyncDeliverMaxResendNum2Blocked();
+ return this.resendCount > maxResendNum2Blocked;
+ }
+
+ public boolean need2Skip() {
+ int maxResendNum2Blocked = ScheduleMessageService.this.defaultMessageStore.getMessageStoreConfig()
+ .getScheduleAsyncDeliverMaxResendNum2Blocked();
+ return this.resendCount > maxResendNum2Blocked * 2;
+ }
+
+ @Override
+ public String toString() {
+ return "PutResultProcess{" +
+ "topic='" + topic + '\'' +
+ ", offset=" + offset +
+ ", physicOffset=" + physicOffset +
+ ", physicSize=" + physicSize +
+ ", delayLevel=" + delayLevel +
+ ", msgId='" + msgId + '\'' +
+ ", autoResend=" + autoResend +
+ ", resendCount=" + resendCount +
+ ", status=" + status +
+ '}';
+ }
+ }
+
+ public enum ProcessStatus {
+ /**
+ * In process, the processing result has not yet been returned.
+ * */
+ RUNNING,
+
+ /**
+ * Put message success.
+ * */
+ SUCCESS,
+
+ /**
+ * Put message exception.
+ * When autoResend is true, the message will be resend.
+ * */
+ EXCEPTION,
+
+ /**
+ * Skip put message.
+ * When the message cannot be looked, the message will be skipped.
+ * */
+ SKIP,
+ }
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/ScheduleMessageServiceTest.java b/store/src/test/java/org/apache/rocketmq/store/ScheduleMessageServiceTest.java
index bfcebd7eb8..1c0451ccc9 100644
--- a/store/src/test/java/org/apache/rocketmq/store/ScheduleMessageServiceTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/ScheduleMessageServiceTest.java
@@ -18,18 +18,36 @@
package org.apache.rocketmq.store;
import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.ThreadFactoryImpl;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.schedule.ScheduleMessageService;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
+import org.junit.Assert;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
public class ScheduleMessageServiceTest {
@@ -40,7 +58,9 @@ public class ScheduleMessageServiceTest {
ConcurrentMap offsetTable = null;
- ScheduleMessageService scheduleMessageService = new ScheduleMessageService((DefaultMessageStore) buildMessageStore());
+ DefaultMessageStore defaultMessageStore = new DefaultMessageStore(buildMessageStoreConfig(),
+ new BrokerStatsManager("simpleTest", true), null, new BrokerConfig());
+ ScheduleMessageService scheduleMessageService = new ScheduleMessageService(defaultMessageStore);
scheduleMessageService.parseDelayLevel();
ConcurrentMap offsetTable1 = new ConcurrentHashMap<>();
@@ -71,7 +91,7 @@ public class ScheduleMessageServiceTest {
}
- private MessageStore buildMessageStore() throws Exception {
+ private MessageStoreConfig buildMessageStoreConfig() throws Exception {
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setMappedFileSizeCommitLog(1024 * 1024 * 10);
messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 1024 * 10);
@@ -79,6 +99,96 @@ public class ScheduleMessageServiceTest {
messageStoreConfig.setMaxIndexNum(100 * 100);
messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
messageStoreConfig.setFlushIntervalConsumeQueue(1);
- return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest", true), null, new BrokerConfig());
+ return messageStoreConfig;
+ }
+
+ @Test
+ public void testHandlePutResultTask() throws Exception {
+ DefaultMessageStore messageStore = mock(DefaultMessageStore.class);
+ MessageStoreConfig config = buildMessageStoreConfig();
+ config.setEnableScheduleMessageStats(false);
+ config.setEnableScheduleAsyncDeliver(true);
+ when(messageStore.getMessageStoreConfig()).thenReturn(config);
+ ScheduleMessageService scheduleMessageService = new ScheduleMessageService(messageStore);
+ scheduleMessageService.parseDelayLevel();
+
+ Field field = scheduleMessageService.getClass().getDeclaredField("deliverPendingTable");
+ field.setAccessible(true);
+ Map> deliverPendingTable =
+ (Map>) field.get(scheduleMessageService);
+
+ field = scheduleMessageService.getClass().getDeclaredField("offsetTable");
+ field.setAccessible(true);
+ ConcurrentMap offsetTable =
+ (ConcurrentMap) field.get(scheduleMessageService);
+ for (int i = 1; i <= scheduleMessageService.getMaxDelayLevel(); i++) {
+ offsetTable.put(i, 0L);
+ }
+
+ int deliverThreadPoolNums = Runtime.getRuntime().availableProcessors();
+ ScheduledExecutorService handleExecutorService = new ScheduledThreadPoolExecutor(deliverThreadPoolNums,
+ new ThreadFactoryImpl("ScheduleMessageExecutorHandleThread_"));
+ field = scheduleMessageService.getClass().getDeclaredField("handleExecutorService");
+ field.setAccessible(true);
+ field.set(scheduleMessageService, handleExecutorService);
+
+ field = scheduleMessageService.getClass().getDeclaredField("started");
+ field.setAccessible(true);
+ AtomicBoolean started = (AtomicBoolean) field.get(scheduleMessageService);
+ started.set(true);
+
+ for (int level = 1; level <= scheduleMessageService.getMaxDelayLevel(); level++) {
+ ScheduleMessageService.HandlePutResultTask handlePutResultTask = scheduleMessageService.new HandlePutResultTask(level);
+ handleExecutorService.schedule(handlePutResultTask, 10L, TimeUnit.MILLISECONDS);
+ }
+
+ MessageExt messageExt = new MessageExt();
+ messageExt.putUserProperty("init", "test");
+ messageExt.getProperties().put(MessageConst.PROPERTY_REAL_QUEUE_ID, "0");
+ when(messageStore.lookMessageByOffset(anyLong(), anyInt())).thenReturn(messageExt);
+ when(messageStore.putMessage(any())).thenReturn(new PutMessageResult(PutMessageStatus.PUT_OK, null));
+
+ int msgNum = 100;
+ int totalMsgNum = msgNum * scheduleMessageService.getMaxDelayLevel();
+ List> putMsgFutrueList = new ArrayList<>(totalMsgNum);
+ for (int level = 1; level <= scheduleMessageService.getMaxDelayLevel(); level++) {
+ for (int num = 0; num < msgNum; num++) {
+ CompletableFuture future = new CompletableFuture<>();
+ ScheduleMessageService.PutResultProcess putResultProcess = scheduleMessageService.new PutResultProcess();
+ putResultProcess = putResultProcess
+ .setOffset(num)
+ .setAutoResend(true)
+ .setFuture(future)
+ .thenProcess();
+ deliverPendingTable.get(level).add(putResultProcess);
+ putMsgFutrueList.add(future);
+ }
+ }
+
+ Collections.shuffle(putMsgFutrueList);
+ Random random = new Random();
+ for (CompletableFuture future : putMsgFutrueList) {
+ PutMessageStatus status;
+ if (random.nextInt(1000) % 2 == 0) {
+ status = PutMessageStatus.PUT_OK;
+ } else {
+ status = PutMessageStatus.OS_PAGECACHE_BUSY;
+ }
+
+ if (random.nextInt(1000) % 2 == 0) {
+ PutMessageResult result = new PutMessageResult(status, null);
+ future.complete(result);
+ } else {
+ future.completeExceptionally(new Throwable("complete exceptionally"));
+ }
+ }
+
+ Thread.sleep(1000);
+ for (int level = 1; level <= scheduleMessageService.getMaxDelayLevel(); level++) {
+ Assert.assertEquals(0, deliverPendingTable.get(level).size());
+ Assert.assertEquals(msgNum, offsetTable.get(level).longValue());
+ }
+
+ scheduleMessageService.shutdown();
}
}