【Apache Rocket】源码解析:消息消费源码解析(监听模式)

2024-01-21 00:36

本文主要是介绍【Apache Rocket】源码解析:消息消费源码解析(监听模式),希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

消费消息源码解析

1. Push模式消费源码解析

相比于生产者,消费者的代码要更为复杂一点,不过也不会太复杂,我们来逐一进行分解。RocketMQ的消费模式一般分为push模式和pull模式;push模式,又可以称为监听模式。

注意:push模式并不是Broker主动去push消息给客户端,本质上还是客户端去pull消息,只不过这个过程客户端是帮你做了而已。实际上还是在内部启了一个PullMessageServieScheduledThread线程专门去向Broker拉取消息。

2. Simple Demo

public static void main(String[] args) throws InterruptedException, MQClientException {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");consumer.subscribe("TopicTest", "*");consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);//wrong time format 2017_0422_221800consumer.setConsumeTimestamp("20181109221800");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.printf("Consumer Started.%n");
}

3. Consumer初始化

consumer init.jpg

  • 首先我们先来看看Consumer是如何初始化的,首先我们要创建一个DefaultMQPushConsumer,再设置对应的参数,然后再注册对应的MessageListener,最后在启动consumer。
  • 启动consumer的过程中,首先看检查相应的配置信息,如consumerGroup、messageModel是否存在,拉取、消费线程的配置情况,拉取队列的大小等等;然后将对应的订阅信息put到rebalance的subscriptionInner中。
  • 根据MessageModel的类型创建并加载对应的offsetStore
    • CLUSTERING:创建并加载RemoteBrokerOffsetStore,CLUSTERING模式的offset存放到Broker端
    • BROADCASTING:创建并加载LocalFileOffsetStore,BROADCASTING模式的offset存在到Client端。
  • 根据MessageListener的类型创建并启动MessageListener,两者的区别可以从字面意思就可以知晓,一个是并发的执行消费逻辑,另一个是顺序的执行消费逻辑。
    • orderly:创建并启动ConsumerMessageOrderlyService和ConsumeMessagePopOrderlyService
    • concurrently:创建并启动ConsumeMessageConcurrentlyService和ConsumeMessagePopConcurrentlyService
  • 将自己注册到MQClientFactory中。
  • 最后在启动MQClientFactory
    • 先检查是否已经获取到了NameServer的地址,没有的话去获取NameServer的地址
    • 再开启request-response的channel
    • 启动schedule任务线程。
      • 定时去获取NameServer的地址
      • 更新对应Topic的路由信息
      • 持久化当前服务的所有的Consumer的offset信息,只持久化LocalFileOffsetStore,即广播消费。
      • 调整线程池线程数量大小,目前concurrent模式和order模式都没有对应的实现。
    • 然后在启动pullService和rebalanceService
    • 最后再启动DefaultMQProducer,为什么要启动Producer?因为在消费监听返回RECONSUMER_LATER后,消息会发送到Broker,这个时候就需要发送者发送对应的RETRY消息。

4. 拉取消息(客户端处理)

当Consumer成功启动后,就开始拉取消息并消费。但是上述过程其实已经启动了拉取、消费线程,具体的执行如下:
pull message.jpg

  • 当start PullMessageService的时候,会启动一个PullMessageServiceScheduledThread线程,这个线程专门负责执行拉取消息的动作,先判断请求模式,POP消息先不做讲解,后面专门讲解POP消息,MessageRequestMode不为POP消息,执行pullMessage方法。
  • 首先是一些前置检查:检查状态是否OK,检查本地缓存的消息的size、byte是否大于阈值,检查无误后,读取offset信息,再到subscribeData中find一个Broker地址,远程请求Broker。
  • 请求Broker之后,等broker处理之后,根据返回的结果执行对应的回调
    • PullCallback.onException:调用executePullRequestLater,延迟相应的时间,执行executePullRequestImmediately,将当前的PullRequest对象put到到messageRequestQueue中。
      • PullMessageServiceScheduledThread线程,会不断地从messageRequestQueue take PullRequest对象,拿到PullRequest对象后,执行PullMessage方法,与上述的pullMessage执行过程一致。
    • PullCallback.onSuccess:先调用processPullResult方法处理broker返回的result,然后再判断状态:
      • FOUND
        • 先将对应的指标加到对应的metrics,判断是否返回了消息
          • 为null或者empty:调用executePullRequestLater
          • 不为null或者empty:先put到msgTreeMap中,然后根据pullInterval判断调用executePullRequestLater或者executePullRequestImmediately。
      • NO_NEW_MSG/NO_MATCHED_MSG:执行correctTagsOffset,更正offset,调用executePullRequestLater
      • OFFSET_ILLEGAL:延迟10s,更新offset信息并持久化到磁盘上(仅针对BROADCASTING模式,CLUSTERING模式的offset保存在broker,所以不需要持久化,这时候客户端对requestBroker进行更新。);然后remove对应的ProcessQueue。
  • 当put完到msgTreeMap中后,执行submitConsumeRequet方法
    • 先判断当前消息条数是否满足消费的阈值,
      • 不满足:执行submitConsumeRequestLater
      • 满足:先把对应的msg、processQueue、messageQueue封装成一个ConsumeRequest对象,丢给消费线程去调用,如果此时抛出rejectedExecutionException,执行submitConsumeRequet
      • ConsumeRequest实现了Runnable接口,所以被消费线程执行时,会执行内部的run方法
        • run:先执行用户自己的消费逻辑并返回结果并处理consumeResult并更新对应的ackIndex
          • 如果ackIndex不满足对应的要求,说明这批消息消费失败,根据messageModel来执行一下逻辑
            • CLUSTERING:执行sendBack,将消息发送给Broker,这就是为什么consumer也需要创建Producer对象的原因;等待sendback时候执行完成,如果执行失败,执行submitConsumeRequestLater,反之则结束。
            • BROADCASTING:输入warn日志即完成。

拉取消息(Broker端处理)

当拉取请求到达Broker端时,执行流程如下图:
pull message for broker.png

  • 拉取请求到达Broker后,通过Broker内部封装的netty,使请求流转到PullMessageProcessor,先进行一些前置检查,如:是否用户当前Broker的拉取权限;当前Broker的主动拉取是否开启;当前ConsumerGroup对应的SubscribeGroup是否存在;当前topic对应的TopicConfig是否存在;校验当前请求参数是否正确等等
  • 然后再查询resetOffsetTable中是否有对应重置之后的offset并且重置操作在Broker进行。
    • 是:把对应的resetSequence set到GetMessageResult中,最后PullMessageResultHandler处理,最后在response给客户端。
    • 否:先getMessageStore(我们以DefaultMessageStore为例,DLedgerMessageStore,后续再详细讲解),在调用messageStore.getMessageAsync()方法,getMessageAsync方法本质上就是用CompletableFuture包了一下getMessage方法。先判断当前topic是否开启了压缩,开启压缩:调用compactionStore.getMessage()(其实本质上还是查CommitLog,所以在流程图中并未体现),拿到ConsumeQueue后,先检查offset是否正确
      • 正确:遍历Mmap直至拿到正确的offsetPy,SizePy,再跟你offsetPy定位到具体的commitlog文件,定位到commitlog文件后,先确定开始拉取的postion(offset%mappedFileSize),把消息写到对应的byteBuffer中。
      • 不正确:修正offset,再把修正之后的offset set到getResult中
  • 上述执行完成后,在PullMessageResultHandler中进行处理
    • 先执行执行之后的回调
    • 如果是广播消费,更新对应广播消费的拉取进度,如果不是广播消费,这步可以忽略
    • 然后把对应的消费进去commit到offsetTable中。
    • 最后在把结果response给客户端。

创作不易:请关注微信公众号:【An前码后】获取更多源码解析,您的关注,就是博主创作的一份动力。

这篇关于【Apache Rocket】源码解析:消息消费源码解析(监听模式)的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/627846

相关文章

PostgreSQL的扩展dict_int应用案例解析

《PostgreSQL的扩展dict_int应用案例解析》dict_int扩展为PostgreSQL提供了专业的整数文本处理能力,特别适合需要精确处理数字内容的搜索场景,本文给大家介绍PostgreS... 目录PostgreSQL的扩展dict_int一、扩展概述二、核心功能三、安装与启用四、字典配置方法

Java设计模式---迭代器模式(Iterator)解读

《Java设计模式---迭代器模式(Iterator)解读》:本文主要介绍Java设计模式---迭代器模式(Iterator),具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,... 目录1、迭代器(Iterator)1.1、结构1.2、常用方法1.3、本质1、解耦集合与遍历逻辑2、统一

深度解析Java DTO(最新推荐)

《深度解析JavaDTO(最新推荐)》DTO(DataTransferObject)是一种用于在不同层(如Controller层、Service层)之间传输数据的对象设计模式,其核心目的是封装数据,... 目录一、什么是DTO?DTO的核心特点:二、为什么需要DTO?(对比Entity)三、实际应用场景解析

Java 线程安全与 volatile与单例模式问题及解决方案

《Java线程安全与volatile与单例模式问题及解决方案》文章主要讲解线程安全问题的五个成因(调度随机、变量修改、非原子操作、内存可见性、指令重排序)及解决方案,强调使用volatile关键字... 目录什么是线程安全线程安全问题的产生与解决方案线程的调度是随机的多个线程对同一个变量进行修改线程的修改操

深度解析Java项目中包和包之间的联系

《深度解析Java项目中包和包之间的联系》文章浏览阅读850次,点赞13次,收藏8次。本文详细介绍了Java分层架构中的几个关键包:DTO、Controller、Service和Mapper。_jav... 目录前言一、各大包1.DTO1.1、DTO的核心用途1.2. DTO与实体类(Entity)的区别1

Java中的雪花算法Snowflake解析与实践技巧

《Java中的雪花算法Snowflake解析与实践技巧》本文解析了雪花算法的原理、Java实现及生产实践,涵盖ID结构、位运算技巧、时钟回拨处理、WorkerId分配等关键点,并探讨了百度UidGen... 目录一、雪花算法核心原理1.1 算法起源1.2 ID结构详解1.3 核心特性二、Java实现解析2.

使用Python绘制3D堆叠条形图全解析

《使用Python绘制3D堆叠条形图全解析》在数据可视化的工具箱里,3D图表总能带来眼前一亮的效果,本文就来和大家聊聊如何使用Python实现绘制3D堆叠条形图,感兴趣的小伙伴可以了解下... 目录为什么选择 3D 堆叠条形图代码实现:从数据到 3D 世界的搭建核心代码逐行解析细节优化应用场景:3D 堆叠图

深度解析Python装饰器常见用法与进阶技巧

《深度解析Python装饰器常见用法与进阶技巧》Python装饰器(Decorator)是提升代码可读性与复用性的强大工具,本文将深入解析Python装饰器的原理,常见用法,进阶技巧与最佳实践,希望可... 目录装饰器的基本原理函数装饰器的常见用法带参数的装饰器类装饰器与方法装饰器装饰器的嵌套与组合进阶技巧

java向微信服务号发送消息的完整步骤实例

《java向微信服务号发送消息的完整步骤实例》:本文主要介绍java向微信服务号发送消息的相关资料,包括申请测试号获取appID/appsecret、关注公众号获取openID、配置消息模板及代码... 目录步骤1. 申请测试系统2. 公众号账号信息3. 关注测试号二维码4. 消息模板接口5. Java测试

解析C++11 static_assert及与Boost库的关联从入门到精通

《解析C++11static_assert及与Boost库的关联从入门到精通》static_assert是C++中强大的编译时验证工具,它能够在编译阶段拦截不符合预期的类型或值,增强代码的健壮性,通... 目录一、背景知识:传统断言方法的局限性1.1 assert宏1.2 #error指令1.3 第三方解决