Shulou(Shulou.com)06/01 Report--

This article mainly introduces how to achieve push consumer sequential consumption in RocketMQ, has a certain reference value, interested friends can refer to, I hope you can learn a lot after reading this article, the following let the editor take you to understand it.

Logical implementation of sequential consumption in class ConsumeMessageOrderlyService, queue needs to be locked in order to achieve the ordering of consumption, including:

Lock the message queue in broker to ensure that the current client owns the queue

Consumer locks the MessageQueue to ensure that the current thread occupies the queue

Consumer locks the ProcessQueue to ensure that the current thread occupies the queue

Locking the message queue on broker is scheduled periodically in ConsumeMessageOrderlyService:

/ / ConsumeMessageOrderlySerivce public void start () {if (MessageModel.CLUSTERING.equals (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel () {this.scheduledExecutorService.scheduleAtFixedRate (new Runnable () {@ Override public void run () {ConsumeMessageOrderlyService.this.lockMQPeriodically ();}}, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS) }} public synchronized void lockMQPeriodically () {if (! this.stopped) {/ / batch lock mq this.defaultMQPushConsumerImpl.getRebalanceImpl () .lockAll ();}} in broker via LOCK_BATCH_MQ request

Consumption request submission in ConsumeMessageOrderlyService:

/ / ConsumeMessageOrderlySerivce public void submitConsumeRequest (final List msgs, final ProcessQueue processQueue, final MessageQueue messageQueue, final boolean dispathToConsume) {if (dispathToConsume) {/ / submit ConsumeRequest, discard the msgs of the input parameter, and get ConsumeRequest consumeRequest = new ConsumeRequest (processQueue, messageQueue); this.consumeExecutor.submit (consumeRequest);}} from ProcessQueue every time

The logic is processed sequentially:

/ / ConsumeMessageOrderlyService.ConsumeRequest 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;} / / 1 Get the lock on MessageQueue 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;) {/ / cyclic processing / maximum processing time of a single ConsumeRequest defaults to 60s 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 () / / 2. Get batchSize messages from ProcessQueue List 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; / /.... Long beginTimestamp = System.currentTimeMillis (); ConsumeReturnType returnType = ConsumeReturnType.SUCCESS; boolean hasException = false; try {/ / 3. Get the lock this.processQueue.getLockConsume (). Lock () on ProcessQueue; if (this.processQueue.isDropped ()) {log.warn ("consumeMessage, the message queue not be able to consume, because it's dropped. {} ", this.messageQueue); break;} / / 4. Push to business processing logic 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 (); / / unlock} /. / / 5. Processing consumption result continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult (msgs, status, context, this);} else {continueConsume = false / / ProcessQueue is empty, stop this push} 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);}

Two main steps are performed in processConsumeResult:

Execute commit () on ProcessQueue to delete the msgs returned by the previous takeMessages from the cache

Update OffsetStore

Thank you for reading this article carefully. I hope the article "how to achieve push consumer Sequential consumption in RocketMQ" shared by the editor will be helpful to everyone. At the same time, I also hope you will support us and pay attention to the industry information channel. More related knowledge is waiting for you to learn!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.