顺序消息原理|学习笔记

简介: 快速学习顺序消息原理

开发者学堂课程【RocketMQ 知识精讲与项目实战(第三阶段)顺序消息原理】学习笔记,与课程紧密联系,让用户快速学习知识。

课程地址:https://developer.aliyun.com/learning/course/704/detail/12505


顺序消息原理

 

顺序消息的原理介绍

顺序消息处理时,入口的位置如下代码,

final long beginTimestamp = System.currentTimerMillis();

Pullcallback pullCallback = new Pullcallback(){

入口在 pullcallback ,一但拿到消息之后,有:

boolean dispatchToConsume =

processQueue.putMessage(pullResult.getMsgFoundList());DefaultMQPushconsumerImpl.this.consumeMessageService.submitConsumeRequest(

pullResult.getMsgFoundList(),

processQueue,

pullRequest.getMessageQueue(),

dispatchToConsume);

如上代码位置,将消息提交到消费者的线程去进行处理,submitConsumeRequest 有两种机制:

image.png

一种是ConsumerMessageConcurrentlyService,一种是

ConsumerMessageOrderlyService。

进入 submitConsumeRequest,

@override

public void submitConsumeRequest(

final List<MessageExt> msgs,

final ProcessQueue processQueue,

final MessageQueue messageQueue,

final boolean dispathToConsume) {

if (dispathToConsume) {

ConsumeRequest consumeRequest = new ConsumeRequest(processQueue,messageQueue);

this.consumeExecutor .submit(consumeRequest);}

}

如上代码位置,将消息提交到消费者的线程,service 在启动时里会有start方法。

public void start(){

if (MessageModel.CLUSTERING.equals(consumeMessageOrderlyService.this.defaultMQPushConsumerImpl.message())){

this.scheduledExecutorService.scheduleAtFixedRate( new Runnable() {

publie void run( ){

consumeMessageOrderlyservice.this.lockMQPeriodically();}

},initialDelay.1000 * 1,ProcessQueue .REBALANCE_LOCK_INTERVL,TimeUnit.MILLISECONDS);}}

这个 start 方法每隔多久去处理信息?

进行以下操作:

image.png点击进入后能够看到下面这行代码:

RVAL =

Long.parseLong(System.getProperty(key:”rocketmq.client.rebalance.lockInterval”, def:”20000”));

所以它会每隔20秒去执提交给它的消息。

打开run 方法,在run方法中处理,顺序消费是一定要基于锁,代码是:final Object objlock =

messageQueueLock.fetchLockObject(this.messageQueue);

synchronized(objLock){

这个锁的作用是当它在消费的时候,其他人不能消费,整个过程是同步的过程。当它拿到锁对象后,会通过 synchronized 将整个消费的过程进行同步的处理。顺序消息关键就是基于同步的方式去处理,但是真正处理业务逻辑和一般的消息没有区别,主要的区别就是加锁的设置。

以上是关于顺序消息的内容。

相关文章
|
消息中间件 NoSQL 关系型数据库
6年高级开发就因这道题少了5K,Kafka如何避免消息重复消费?
一个6年工作经验的小伙伴,被问到这样一个问题,说Kafka是如何避免消息重复消费的?面试完之后,这位小伙伴来找到我,希望我能给一个思路。今天,我给大家分享一下我的思路。
165 1
|
7月前
|
消息中间件 存储 Kafka
几种 MQ 顺序消息的实现方式
几种 MQ 顺序消息的实现方式
|
存储
什么是队列,如何实现?
什么是队列,如何实现?
116 0
什么是队列,如何实现?
|
消息中间件 Java uml
5张图带你理解 RocketMQ 顺序消息实现机制
5张图带你理解 RocketMQ 顺序消息实现机制
708 1
5张图带你理解 RocketMQ 顺序消息实现机制
|
消息中间件 RocketMQ 开发者
顺序消息原理|学习笔记
快速学习顺序消息原理
顺序消息原理|学习笔记
|
消息中间件 存储 RocketMQ
【视频】顺序消息| 学习笔记
快速学习【视频】顺序消息
【视频】顺序消息| 学习笔记
|
消息中间件 RocketMQ 开发者
顺序消息分析|学习笔记
快速学习顺序消息分析
顺序消息分析|学习笔记
|
消息中间件 Java 开发者
消息类型-顺序消息|学习笔记
快速学习消息类型-顺序消息
123 0
消息类型-顺序消息|学习笔记
|
存储 消息中间件 Linux
多类型业务消息专题-顺序消息 | 学习笔记
快速学习多类型业务消息专题-顺序消息
多类型业务消息专题-顺序消息 | 学习笔记
|
消息中间件 Java
浅析什么是顺序消息?
在日常开发中,或多或少的都遇到过顺序消息的使用场景,那么为什么要使用顺序消息呢?如果不使用顺序消息会有什么问题吗?市面上的消息中间件又是如何实现的呢?
219 0
浅析什么是顺序消息?
下一篇
DataWorks