sequence
This paper mainly studies The ConsumeMode.ORDERLY of RocketMQ
ConsumeMode.ORDERLY
Rocketmq – spring – the boot – 2.0.4 – sources. The jar! /org/apache/rocketmq/spring/annotation/ConsumeMode.java
public enum ConsumeMode {
/**
* Receive asynchronously delivered messages concurrently
*/
CONCURRENTLY,
/**
* Receive asynchronously delivered messages orderly. one queue, one thread
*/
ORDERLY
}
Copy the code- ConsumeMode defines two enumerated values of CONCURRENTLY and ORDERLY
DefaultRocketMQListenerContainer
Rocketmq – spring – the boot – 2.0.4 – sources. The jar! /org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
public class DefaultRocketMQListenerContainer implements InitializingBean,
RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware {
//......
private void initRocketMQPushConsumer() throws MQClientException {
Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required");
Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
Assert.notNull(nameServer, "Property 'nameServer' is required");
Assert.notNull(topic, "Property 'topic' is required");
RPCHook rpcHook = RocketMQUtil.getRPCHookByAkSk(applicationContext.getEnvironment(),
this.rocketMQMessageListener.accessKey(), this.rocketMQMessageListener.secretKey());
boolean enableMsgTrace = rocketMQMessageListener.enableMsgTrace();
if (Objects.nonNull(rpcHook)) {
consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook, new AllocateMessageQueueAveragely(),
enableMsgTrace, this.applicationContext.getEnvironment().
resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
consumer.setVipChannelEnabled(false);
consumer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook, consumerGroup));
} else {
log.debug("Access-key or secret-key not configure in " + this + ".");
consumer = new DefaultMQPushConsumer(consumerGroup, enableMsgTrace,
this.applicationContext.getEnvironment().
resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
}
String customizedNameServer = this.applicationContext.getEnvironment().resolveRequiredPlaceholders(this.rocketMQMessageListener.nameServer());
if(customizedNameServer ! = null) { consumer.setNamesrvAddr(customizedNameServer); }else {
consumer.setNamesrvAddr(nameServer);
}
if(accessChannel ! = null) { consumer.setAccessChannel(accessChannel); } consumer.setConsumeThreadMax(consumeThreadMax);if (consumeThreadMax < consumer.getConsumeThreadMin()) {
consumer.setConsumeThreadMin(consumeThreadMax);
}
consumer.setConsumeTimeout(consumeTimeout);
switch (messageModel) {
case BROADCASTING:
consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
break;
case CLUSTERING:
consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
break;
default:
throw new IllegalArgumentException("Property 'messageModel' was wrong.");
}
switch (selectorType) {
case TAG:
consumer.subscribe(topic, selectorExpression);
break;
case SQL92:
consumer.subscribe(topic, MessageSelector.bySql(selectorExpression));
break;
default:
throw new IllegalArgumentException("Property 'selectorType' was wrong.");
}
switch (consumeMode) {
case ORDERLY:
consumer.setMessageListener(new DefaultMessageListenerOrderly());
break;
case CONCURRENTLY:
consumer.setMessageListener(new DefaultMessageListenerConcurrently());
break;
default:
throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
}
if(rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) { ((RocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(consumer); }}}Copy the code- InitRocketMQPushConsumer method for consumeMode for ORDERLY setting messageListener for DefaultMessageListenerOrderly
DefaultMessageListenerOrderly
Rocketmq – spring – the boot – 2.0.4 – sources. The jar! /org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
@SuppressWarnings("unchecked")
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
for (MessageExt messageExt : msgs) {
log.debug("received msg: {}", messageExt);
try {
long now = System.currentTimeMillis();
rocketMQListener.onMessage(doConvertMessage(messageExt));
long costTime = System.currentTimeMillis() - now;
log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
} catch (Exception e) {
log.warn("consume message failed. messageExt:{}", messageExt, e);
context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis);
returnConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; }}returnConsumeOrderlyStatus.SUCCESS; }}Copy the code- DefaultMessageListenerOrderly MessageListenerOrderly interface is achieved, the consumeMessage methods traverse MSGS, Then each call rocketMQListener. OnMessage (doConvertMessage (messageExt)), If the loop can perform the context on the abnormal captured. SetSuspendCurrentQueueTimeMillis (suspendCurrentQueueTimeMillis), And then return ConsumeOrderlyStatus SUSPEND_CURRENT_QUEUE_A_MOMENT
ConsumeMessageOrderlyService
Rocketmq – the client – 4.6.0 – sources jar! /org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java
class ConsumeRequest implements Runnable {
private final ProcessQueue processQueue;
private final MessageQueue messageQueue;
public ConsumeRequest(ProcessQueue processQueue, MessageQueue messageQueue) {
this.processQueue = processQueue;
this.messageQueue = messageQueue;
}
public ProcessQueue getProcessQueue() {
return processQueue;
}
public MessageQueue getMessageQueue() {
return messageQueue;
}
@Override
public void run() {
if (this.processQueue.isDropped()) {
log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
return;
}
final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
synchronized (objLock) {
if(MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel()) || (this.processQueue.isLocked() && ! this.processQueue.isLockExpired())) { final long beginTime = System.currentTimeMillis();for (boolean continueConsume = true; continueConsume; ) {
if (this.processQueue.isDropped()) {
log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
break;
}
if(MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel()) && ! this.processQueue.isLocked()) { log.warn("the message queue not locked, so consume later, {}", this.messageQueue);
ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
break;
}
if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
&& this.processQueue.isLockExpired()) {
log.warn("the message queue lock expired, so consume later, {}", this.messageQueue);
ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
break;
}
long interval = System.currentTimeMillis() - beginTime;
if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) {
ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
break;
}
final int consumeBatchSize =
ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
List<MessageExt> msgs = this.processQueue.takeMessags(consumeBatchSize);
defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
if(! msgs.isEmpty()) { final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue); ConsumeOrderlyStatus status = null; ConsumeMessageContext consumeMessageContext = null;if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
consumeMessageContext = new ConsumeMessageContext();
consumeMessageContext
.setConsumerGroup(ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumerGroup());
consumeMessageContext.setNamespace(defaultMQPushConsumer.getNamespace());
consumeMessageContext.setMq(messageQueue);
consumeMessageContext.setMsgList(msgs);
consumeMessageContext.setSuccess(false);
// init the consume context type
consumeMessageContext.setProps(new HashMap<String, String>());
ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
}
long beginTimestamp = System.currentTimeMillis();
ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
boolean hasException = false;
try {
this.processQueue.getLockConsume().lock();
if (this.processQueue.isDropped()) {
log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}",
this.messageQueue);
break;
}
status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
} catch (Throwable e) {
log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}",
RemotingHelper.exceptionSimpleDesc(e),
ConsumeMessageOrderlyService.this.consumerGroup,
msgs,
messageQueue);
hasException = true;
} finally {
this.processQueue.getLockConsume().unlock();
}
if (null == status
|| ConsumeOrderlyStatus.ROLLBACK == status
|| ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
log.warn("consumeMessage Orderly return not OK, Group: {} Msgs: {} MQ: {}",
ConsumeMessageOrderlyService.this.consumerGroup,
msgs,
messageQueue);
}
long consumeRT = System.currentTimeMillis() - beginTimestamp;
if (null == status) {
if (hasException) {
returnType = ConsumeReturnType.EXCEPTION;
} else {
returnType = ConsumeReturnType.RETURNNULL; }}else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) {
returnType = ConsumeReturnType.TIME_OUT;
} else if (ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
returnType = ConsumeReturnType.FAILED;
} else if (ConsumeOrderlyStatus.SUCCESS == status) {
returnType = ConsumeReturnType.SUCCESS;
}
if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
consumeMessageContext.getProps().put(MixAll.CONSUME_CONTEXT_TYPE, returnType.name());
}
if (null == status) {
status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
}
if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
consumeMessageContext.setStatus(status.toString());
consumeMessageContext
.setSuccess(ConsumeOrderlyStatus.SUCCESS == status || ConsumeOrderlyStatus.COMMIT == status);
ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
}
ConsumeMessageOrderlyService.this.getConsumerStatsManager()
.incConsumeRT(ConsumeMessageOrderlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);
continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
} else {
continueConsume = false; }}}else {
if (this.processQueue.isDropped()) {
log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
return; } ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100); }}}}Copy the code- The ConsumeRequest implements the Runnable interface, and its constructor requires that the processQueue and messageQueue arguments be passed in; The run method firstly messageQueueLock. FetchLockObject (enclosing messageQueue) get objLock, synchronized the objLock conducting follow-up after operation
- And impersonal messageModel messageModel. –
(this.processQueue.isLocked() && ! this.processQueue.isLockExpired())If not, execute tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100) - After through processQueue takeMessags (consumeBatchSize), and then execute processQueue. GetLockConsume (). The lock (), To perform a messageListener. ConsumeMessage (Collections. UnmodifiableList (MSGS), the context), Finally in the finally processQueue. GetLockConsume (). Unlock (), After through ConsumeMessageOrderlyService. This. ProcessConsumeResult (MSGS, status and context, this) processing ConsumeOrderlyStatus
ProcessQueue
Rocketmq – the client – 4.6.0 – sources jar! /org/apache/rocketmq/client/impl/consumer/ProcessQueue.java
public class ProcessQueue {
//......
public List<MessageExt> takeMessags(final int batchSize) {
List<MessageExt> result = new ArrayList<MessageExt>(batchSize);
final long now = System.currentTimeMillis();
try {
this.lockTreeMap.writeLock().lockInterruptibly();
this.lastConsumeTimestamp = now;
try {
if(! this.msgTreeMap.isEmpty()) {for (int i = 0; i < batchSize; i++) {
Map.Entry<Long, MessageExt> entry = this.msgTreeMap.pollFirstEntry();
if(entry ! = null) { result.add(entry.getValue()); consumingMsgOrderlyTreeMap.put(entry.getKey(), entry.getValue()); }else {
break; }}}if (result.isEmpty()) {
consuming = false;
}
} finally {
this.lockTreeMap.writeLock().unlock();
}
} catch (InterruptedException e) {
log.error("take Messages exception", e);
}
returnresult; } / /... }Copy the code- The takeMessags method starts with locktreemap.writelock ().lockInterruptibly(), Then execute msgTreeMap. PollFirstEntry () and consumingMsgOrderlyTreeMap. Put (entry. The getKey (), entry. The getValue ()); Finally execute locktreemap.writelock ().unlock()
summary
DefaultMessageListenerOrderly MessageListenerOrderly interface is achieved, the consumeMessage methods traverse MSGS, Then each call rocketMQListener. OnMessage (doConvertMessage (messageExt)), If the loop can perform the context on the abnormal captured. SetSuspendCurrentQueueTimeMillis (suspendCurrentQueueTimeMillis), And then return ConsumeOrderlyStatus SUSPEND_CURRENT_QUEUE_A_MOMENT
doc
- ConsumeMessageOrderlyService