顺序消息原理|学习笔记

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

开发者学堂课程【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 将整个消费的过程进行同步的处理。顺序消息关键就是基于同步的方式去处理,但是真正处理业务逻辑和一般的消息没有区别,主要的区别就是加锁的设置。

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

相关文章
|
8月前
|
安全 数据安全/隐私保护 Windows
如何在Windows 10系统中查看已连接WiFi密码-亲测可用-优雅草卓伊凡
如何在Windows 10系统中查看已连接WiFi密码-亲测可用-优雅草卓伊凡
435 15
如何在Windows 10系统中查看已连接WiFi密码-亲测可用-优雅草卓伊凡
|
监控 微服务 Python
微服务的故障恢复与弹性设计
【8月更文第29天】在微服务架构中,由于服务间的相互依赖,任何单点故障都可能导致整个系统崩溃。因此,设计具备高可用性和弹性的微服务系统至关重要。本文将探讨如何通过重试机制、断路器和超时设置等策略来增强系统的容错能力和恢复能力。
550 2
|
Linux iOS开发 网络架构
如何使用 Ping 命令监测网络丢包情况?
如何使用 Ping 命令监测网络丢包情况?
10898 48
|
存储 缓存 应用服务中间件
|
SQL 数据处理 开发工具
实时计算 Flink版产品使用合集之PyFlink的优势是什么
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
181 1
|
算法 调度
FreeRTOS入门教程(互斥锁的概念和函数使用)
FreeRTOS入门教程(互斥锁的概念和函数使用)
978 0
|
数据挖掘 API 网络安全
如何使用Google Compute Engine入门指南快速创建和配置您的云虚拟机实例
Google Compute Engine(GCE)是 Google Cloud Platform(GCP)的一部分,它为用户提供了高度可扩展的虚拟机实例,用于运行各种工作负载,从简单的网站托管到复杂的数据分析。本文将引导您完成使用 Google Compute Engine 的入门过程,以便您可以轻松开始使用这一强大的云计算平台。
1673 0
|
编解码 计算机视觉 C++
FFmpeg 的使用与Docker安装流媒体服务器
FFmpeg 的使用与Docker安装流媒体服务器
1241 0
|
Java Spring
Java——spring boot打的jar包如何引用外部application.properties
Java——spring boot打的jar包如何引用外部application.properties
|
机器学习/深度学习 人工智能 自然语言处理
超越Transformer,清华、字节大幅刷新并行文本生成SoTA性能|ICML 2022
超越Transformer,清华、字节大幅刷新并行文本生成SoTA性能|ICML 2022
332 0
超越Transformer,清华、字节大幅刷新并行文本生成SoTA性能|ICML 2022