Apache RocketMQ 从入门到实战.pdf

作者简介作者简介作者简介丁威，RocketMQ 技术内幕作者，RocketMQ 官方社区优秀布道师，荣获CSDN2020 博客之星亚军；担任中通快递研发中心资深架构师，维护中间件兴趣圈公众号，主打成体系剖析 Java 主流中间件，尝试从源码分析、架构设计、实战、故障分析等维度深刻揭晓中间件技术，已覆盖 RocketMQ、Dubbo、Sentienl、Kafka、Canal、MyCat、ElasticJob、ElasticSearch 等。推荐人及推荐序推荐人推荐人杜恒，Apache RocketMQ PMC Member/committer，Linux OpenMessaging TSCMember，目前负责 RocketMQ 专有云商业化以及开源技术生态构建。具有多年分布式系统、中间件研究及工程经验。目前对分布式中间件、K8s、微服务、物联网、Serverless 感兴趣。推荐序推荐序Apache RocketMQ 作为一款高吞吐，抗万亿消息堆积的云原生消息平台，目前已经被国内 75%以上互联网、金融等公司所采用，逐渐成为企业 IT 架构的核心基础设施。丁威老师作为资深架构师，在分布式架构、存储方面功底深厚，目前在企业内部负责着日均千亿级消息流转的 RocketMQ 集群。本书不仅由浅入深的介绍了 RocketMQ 的架构与实现，而且包含了多年线上超大规模集群开发运维经验的总结，通过本书不仅能够掌握分布式消息平台的设计原理，对线上疑难问题排查分析、性能调优与架构设计也大有帮助。目录开篇：我的另一种参与 RocketMQ 开源社区的方式61.1 RocketMQ 核心概念扫盲篇101.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 true181.3 实战：RocketMQ 学习环境搭建指南篇281.4 RocketMQ HA 核心工作机制391.5 踩坑记：rocketmq-console 消费 TPS 为 0，但消息积压数却在降低是个什么“鬼”491.6 RocketMQ 一个新的消费组初次启动时从何处开始消费呢？641.7 一次 RocketMQ 进程自动退出排查经验分享781.8 RocketMQ 主题扩分片后遇到的坑821.9 RocketMQ 消息发送 system busy、broker busy 原因分析与解决方案坑911.10 再谈 RocketMQ broker busy1041.11 从年末生产故障解锁 RocketMQ 集群部署的最佳实践1081.12 RocketMQ 一行代码造成大量消息丢失1151.13 RocketMQ DLedger 多副本即主从切换实战1211.14 RocketMQ msgId 与 offsetMsgId 释疑1311.15 RocketMQ ACL 使用指南1411.16 RocketMQ 消息轨迹-设计篇1511.17 消息发送常见问题与解决方案155本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。开篇：我的另一种参与 RocketMQ 开源社区的方式开篇：我的另一种参与 RocketMQ 开源社区的方式RocketMQ 为什么性能高效，到底运用了什么“厉害”的技术？RocketMQ 如何实现刷盘（可以类比一下数据库方面的刷盘、redo、undo 日志）？RocketMQ 文件存储设计理念、基于文件的 Hash 索引是怎么实现的?定时消息、消息过滤等实现原理。如何进行网络编程（Netty 实战）？下定决心后便开始了我的源码分析 RocketMQ 之旅，大概在 4 个多月的时间中连续发表了 30 余篇文章，从 Nameserver、消息发送高可用设计、消息存储、消息消费、消息过滤、事务消息等各个方面对其进行了体系化的剖析，边写边分享，边分享边传播，终于得到了机械工业出版社华章分社的杨福川老师的认可，邀请我出书。在杨老师和张工的帮助与指点下，经过将近半年的努力，书稿基本完稿。由于我当时是一位名不经传的新人，按照出版行业的惯例，需要找一些该领域内专家大牛帮忙做序或写写推荐语。当时我也是初生牛犊不怕虎，蹦出了一个非常大胆的想法，是不是可以联系到RocketMQ 官方的一些大佬，最终我直接锁定了 RocketMQ 创始人冯嘉大神，希望他能帮我作序推荐，令人惊喜的是冯嘉大神非常平易见人，得知我的来意后，他说了这样一句话：“我是非常愿意为写书的朋友作序，但需要评估一下书稿的质量，如果质量 OK，非常愿意效劳”。我备受鼓舞，在和出版社初步沟通后，将试读稿件再加上消息存储整章的内容发给冯嘉大神后，经冯嘉大神认真审稿后，决定帮忙推荐作序，真的非常受鼓舞。随着RocketMQ 技术内幕一书的正式出版上市，并得到广大读者朋友的认可，与官方的联系也越来越多，后面在 RocketMQ 中国社区负责人青峰大佬的筹备下，我还参与了 RocketMQ 官方社区的源码解析直播活动、官方文档审稿等工作，并在社区得到了不错的反响。说到这里大家是不是觉得非常奇怪，是不是都认为你只是在写文章，写书，没有真正参与开源社区呀，没有贡献代码，这个算哪门子参与开源社区？其实我一开始连我自己也没有意识到我正在参与一个开源项目，直到我在冯嘉大神为我写的序言中给了我一个新的称号：RocketMQ 布道师，从而才真正了解到参与开源的另外一种方式：做一个开源项目的传播者，让更多人更容易的应用它，即降低大众对它的使用门槛。本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。开篇：我的另一种参与 RocketMQ 开源社区的方式开篇：我的另一种参与 RocketMQ 开源社区的方式接下来我们回到本节的主题，那如何参与一个开源项目呢？在参与一个开源项目之前，我觉得第一个最基本的步骤还是要打牢基础，这里的基础至少要包括 JAVA 集合、JAVA 并发（JUC)这两项，只是最最基本的，至少要阅读其源码，理解其设计理念，至于 NIO，Netty 这些可以后续在需要使用时再去专门学习，有针对性的学习，有使用需求，或许学习动力更强劲，学习效率更高效。当具备一定的基础后，如何从零开始参与进开源项目呢？通常有如下几个方法：看看官方文档，特别是设计手册，从整体上把握其设计理念。写写源码分析类文章，从整体上把控这个框架，这个花费时间较多，如果框架正在起步。阶段，不建议该方法；如果框架比较成熟，非常建议采用该方法。尝试看看开源项目中的 issues，看能不能解决，从问题入手，快速融入该项目。尝试谢谢单元测试用例，测试驱动开发，借此学习该框架。后面的事情就是坚持不懈，朝着目标不断前进，中途可以放慢速度，但千万别放弃，因为只有坚持，才能胜利，只要前进，就离目标更近。参与开源，一个最基本的条件是拥有大量的连续时间，想要成为一个开源框架的Committer,唯有坚持不懈，持续投入，持续产出。最后再次感谢 RocketMQ 社区对我的认可，我会尽努力做出更大的贡献，也希望广大读者朋友们，积极参与开源社区，贡献一份自己的力量，同事打造自身影响力，助力职场步步高升。本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.1 RocketMQ 核心概念扫盲篇1.1 RocketMQ 核心概念扫盲篇3.3.ClientClient消息客户端，包括 Producer(消息发送者)和 Consumer(消费消费者)客户端在同一时间只会连接一台 nameserver，只有在连接出现异常时才会向尝试连接另外一台。客户端每隔 30s 向 Nameserver 发起 topic 的路由信息查询。温馨提示：Nameserver 是在内存中存储 Topic 的路由信息，持久化 Topic 路由信息的地方是在 Broker 中，即$ROCKETMQ_HOME/store/config/topics.json。在 RocketMQ4.5.0 版本后引入了多副本机制，即一个复制组（m-s）可以演变为基于 raft 协议的复制组，复制组内部使用 raft 协议保证 broker 节点数据的强一致性，该部署架构在金融行业用的比较多。二二、消息订阅模型、消息订阅模型在 RocketMQ 的消息消费模式采用的是发布与订阅模式。topic：一类消息的集合，消息发送者将一类消息发送到一个主题中，例如订单模块将订单发送到 order_topic 中，而用户登录时，将登录事件发送到 user_login_topic 中。consumegroup：消息消费组，一个消费单位的“群体”，消费组首先在启动时需要订阅需要消费的 topic。一个 topic 可以被多个消费组订阅，同样一个消费组也可以订阅多个主题。一个消费组拥有多个消费者。术语解释起来有点枯燥晦涩，接下来我举例来阐述。术语解释起来有点枯燥晦涩，接下来我举例来阐述。例如我们在开发一个订单系统，其中有一个子系统：order-service-app，在该项目中会创建一个消费组 order_consumer 来订阅 order_topic，并且基于分布式部署，order-service-app 的部署情况如下：本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.1 RocketMQ 核心概念扫盲篇1.1 RocketMQ 核心概念扫盲篇在在 MQMQ 领域有一个不成文的约定：同一个消费者同一时间可以分配多个队列，但一个领域有一个不成文的约定：同一个消费者同一时间可以分配多个队列，但一个队列同一时间只会分配给一个消费者。队列同一时间只会分配给一个消费者。RocketMQ 提供了众多的队列负载算法，其中最常用的两种平均分配算法。AllocateMessageQueueAveragely平均分配AllocateMessageQueueAveragelyByCircle轮流平均分配为了说明这两种分配算法的分配规则，现在对 16 个队列，进行编号，用 q0q15 表示，消费者用 c0c2 表示。AllocateMessageQueueAveragely 分配算法的队列负载机制如下：c0：q0 q1 q2 q3 q4 q5c1:q6 q7 q8 q9 q10c2:q11 q12 q13 q14 q15其算法的特点是用总数除以消费者个数，余数按消费者顺序分配给消费者，故 c0 会多分配一个队列，而且队列分配是连续的。AllocateMessageQueueAveragelyByCircle 分配算法的队列负载机制如下：c0：q0 q3 q6 q9 q12 q15c1:q1q4 q7 q10 q13c2:q2q5 q8 q11 q14本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.1 RocketMQ 核心概念扫盲篇1.1 RocketMQ 核心概念扫盲篇上述整个过程无需应用程序干预，由 RocketMQ 完成。大概的做法就是将将原先分配给自己但这次不属于的队列进行丢弃，新分配的队列则创建新的拉取任务。3.3.消费进度消费进度消费者消费一条消息后需要记录消费的位置，这样在消费端重启的时候，继续从上一次消费的位点开始进行处理新的消息。在 RocketMQ 中，消息消费位点的存储是以消费组为单位的。集群模式集群模式下，消息消费进度存储在 broker 端，$ROCKETMQ_HOME/store/config/consumerOffset.json 是其具体的存储文件，其中内容截图如下：可见消费进度的 Key 为：topicconsumeGroup，然后每一个队列一个偏移量。广播模式广播模式的消费进度文件存储在用户的主目录，默认文件全路劲名：$USER_HOME/.rocketmq_offsets。4.4.消费模型消费模型RocketMQ 提供了并发消费、顺序消费两种消费模型。并发消费并发消费：对一个队列中消息，每一个消费者内部都会创建一个线程池，对队列中的消息多线程处理，即偏移量大的消息比偏移量小的消息有可能先消费。顺序消费顺序消费：在某一项场景，例如 MySQL binlog 场景，需要消息按顺序进行消费。在RocketMQ 中提供了基于队列的顺序消费模型，即尽管一个消费组中的消费者会创建一个多线程，但针对同一个 Queue，会加锁。温馨提示：并发消费模型中，消息消费失败默认会重试 16 次，每一次的间隔时间不一样；而顺序消费，如果一条消息消费失败，则会一直消费，直到消费成功。故在顺序消费的本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.1 RocketMQ 核心概念扫盲篇1.1 RocketMQ 核心概念扫盲篇RocketMQ 目前支持指定级别的延迟，其延迟级别如下：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h五五、消息过滤、消息过滤消息过滤是指消费端可以根据某些条件对一个 topic 中的消息进行过滤，即只消费一个主题下满足过滤条件的消息。RocketMQ 目前主要的过滤机制是基于 tag 的过滤与基于消息属性的过滤，基于消息属性的过滤支持 SQL92 表达式，对消息进行过滤。六六、小结、小结本文的主要目的是介绍 RocketMQ 常见的术语，例如 nameserver、broker、主题、消费组、消费者、队列负载算法、队列重平衡机制、并发消费、顺序消费、消费进度存储、定时消息、事务消息、消息过滤等基本概念，为后续的实战系列打下坚实基础。从下一篇开始，将正式开始 RocketMQ 之旅，开始学习消息发送。本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 true1.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 true默认读写队列的个数为 4。我们再来看一下 RocketMQ 默认 topic 的路由信息截图如下：从图中可以默认 Topic 的路由信息为 broker-a、broker-b 上各 8 个队列。二二、思考、思考默认 Topic 的路由信息是如何创建的？Topic 的路由信息是存储在哪里？Nameserver？broker?RocketMQ Topic 默认队列个数。三三、原理、原理1.1.RocketMQRocketMQ 基本路由规则基本路由规则本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 true1.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 true但问题就来了，默认 Topic 在集群的每一台 Broker 上创建 8 个队列，那问题来了，为啥新创建的 Topic 只在一个 Broker 上创建 4 个队列？2.2.探究探究 autoCreateTopicEnableautoCreateTopicEnable 机制机制默认默认 TopicTopic 路由创建时机路由创建时机温馨提示：本文不会详细跟踪整个整个源码创建过程，只会点出代码的关键入口点，如想详细了解 NameServer 路由消息、消息发送高可用的实现原理，建议查阅笔者的书籍RocketMQ 技术内幕第二、三章。Step1：在 Broker 启动流程中，会构建 TopicConfigManager 对象，其构造方法中首先会判断是否开启了允许自动创建主题，如果启用了自动创建主题，则向 topicConfigTable 中添加默认主题的路由信息。TopicConfigManager 构造方法：本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。211.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 true备注：该 topicConfigTable 中所有的路由信息，会随着 Broker 向 Nameserver 发送心跳包中，Nameserver 收到这些信息后，更新对应 Topic 的路由信息表。注意：BrokerConfig 的 defaultTopicQueueNum 默认为 8。两台 Broker 服务器都会运行上面的过程，故最终 Nameserver 中关于默认主题的路由信息中，会包含两个Broker 分别各 8 个队列信息。Step2：生产者寻找路由信息生产者首先向 NameServer 查询路由信息，由于是一个不存在的主题，故此时返回的路由信息为空，RocketMQ 会使用默认的主题再次寻找，由于开启了自动创建路由信息，NameServer 会向生产者返回默认主题的路由信息。然后从返回的路由信息中选择一个队列（默认轮询）。消息发送者从 Nameserver 获取到默认的 Topic 的队列信息后，队列的个数会改变吗？答案是会的，其代码如下：MQClientInstance#updateTopicRouteInfoFromNameServer本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 true22温馨提示：消息发送者在到默认路由信息时，其队列数量，会选择 DefaultMQProducer#defaultTopicQueueNums 与 Nameserver 返回的的队列数取最小值，DefaultMQProducer#defaultTopicQueueNums 默认值为 4，故自动创建的主题，其队列数量默认为 4。Step3：发送消息DefaultMQProducerImpl#sendKernelImpl在消息发送时的请求报文中，设置默认 topic 名称，消息发送 topic 名称，使用的队列数量为 DefaultMQProducer#defaultTopicQueueNums，即默认为 4。Step4：Broker 端收到消息后的处理流程服务端收到消息发送的处理器为：SendMessageProcessor，在处理消息发送时，会调用 super.msgCheck 方法：AbstractSendMessageProcessor#msgCheck本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 true1.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 true现象分析现象分析经过上面自动创建路由机制的创建流程，我们可以比较容易的分析得出如下结论：因为开启了自动创建路由信息，消息发送者根据 Topic 去 NameServer 无法得到路由信息，但接下来根据默认 Topic 从 NameServer 是能拿到路由信息(在每个 Broker 中，存在 8 个队列)，因为两个 Broker 在启动时都会向 NameServer 汇报路由信息。此时消息发送者缓存的路由信息是 2 个 Broker，每个 Broker 默认 4 个队列（原因见 3.2.1:Step2 的分析）。消息发送者然后按照轮询机制，发送第一条消息选择(broker-a 的本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 true26messageQueue:0)，向 Broker 发送消息，Broker 服务器在处理消息时，首先会查看自己的路由配置管理器(TopicConfigManager)中的路由信息，此时不存在对应的路由信息，然后尝试查询是否存在默认 Topic 的路由信息，如果存在，说明启用了 autoCreateTopicEnable，则在 TopicConfigManager 中创建新 Topic 的路由信息，此时存在与 Broker服务端的内存中，然后本次消息发送结束。此时，在 NameServer 中还不存在新创建的Topic 的路由信息。这里有三个关键点：1.启用 autoCreateTopicEnable 创建主题时，在 Broker 端创建主题的时机为，消息生产者往 Broker 端发送消息时才会创建。2.然后 Broker 端会在一个心跳包周期内，将新创建的路由信息发送到 NameServer，于此同时，Broker 端还会有一个定时任务，定时将内存中的路由信息，持久化到Broker 端的磁盘上。3.消息发送者会每隔 30s 向 NameServer 更新路由信息，如果消息发送端一段时间内未发送消息，就不会有消息发送集群内的第二台 Broker，那么 NameServer 中新创建的 Topic 的路由信息只会包含 Broker-a，然后消息发送者会向 NameServer 拉取最新的路由信息，此时就会消息发送者原本缓存了 2 个 broker 的路由信息，将会变为一个 Broker 的路由信息，则该 Topic 的消息永远不会发送到另外一个 Broker，就出现了上述现象。原因就分析到这里了，现在我们还可以的大胆假设，开启 autoCreateTopicEnable机制，什么情况会在两个 Broker 上都创建队列，其实，我们只需要连续快速的发送 9 条消息，就有可能在 2 个 Broker 上都创建队列，验证代码如下：public static void main(String args)throws MQClientException,InterruptedExceptionDefaultMQProducer producer=new DefaultMQProducer(please_rename_unique_group_name);producer.setNamesrvAddr(127.0.0.1:9876);producer.start();for(int i=0;i 1.2 生产环境中，autoCreateTopicEnable 为什么不能设置为 trueSendResult sendResult=producer.send(msg);System.out.printf(%s%n,sendResult);catch(Exception e)e.printStackTrace();Thread.sleep(1000);producer.shutdown();其路由信息如下，符合预期。本文就分析到这里，如果喜欢这篇文章，希望大家帮忙点赞，转发，谢谢你们，同时大家也可以给作者留言在使用 RocketMQ 的过程中遇到的疑难杂症，与作者互动。本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.3 实战：RocketMQ 学习环境搭建指南篇1.3 实战：RocketMQ 学习环境搭建指南篇其中 conf 文件夹存放的是 RocketMQ 的配置文件，提供了各种部署结构的示例配置。例如 2m-2s-async 是 2 主 2 从异步复制的配置示例；2m-noslave 是 2 主的示例配置。由于本文主要是搭建一个学习环境，故采取的部署架构为 1 主的部署架构，关于生产环境下如何搭建 RocketMQ 集群、如何调优参数将在该专栏的后续文章中专门介绍。Step3：修改 Nameserver jvm 参数cd binvi runserver.sh#定位到如下代码JAVA_OPT=$JAVA_OPT-server-Xms4g-Xmx4g-Xmn2g-XX:MetaspaceSize=128m-XX:MaxMetaspaceSize=320m#修改-Xms-Xmx-Xmn参数JAVA_OPT=$JAVA_OPT-server-Xms512M-Xmx512M-Xmn256M-XX:MetaspaceSize=128m-XX:MaxMetaspaceSize=320m温馨提示：这里修改 JVM 参数主要目的是个人学习电脑内存不够，默认NameServer 会占用 4G。Step4：启动 nameservernohup./mqnamesrv&查看$user_home/logs/rocketmqlogs/namesrv.log 日志文件，如果输出结果如下图所示即表示启动成功。本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.3 实战：RocketMQ 学习环境搭建指南篇1.3 实战：RocketMQ 学习环境搭建指南篇cd binvi runbroker.sh#修改如下配置(配置前)JAVA_OPT=$JAVA_OPT-server-Xms8g-Xmx8g-Xmn4g#配置后JAVA_OPT=$JAVA_OPT-server-Xms1g-Xmx1g-Xmn512mStep7：启动 brokercd binnohup./mqbroker-c./conf/broker.conf&查看$user_home/logs/rocketmqlogs/broker.log，如果输出结果如下图所示表示启动成功。经过上面的步骤，就成功在 Linux 环境上安装了 RocketMQ Nameserver 服务器与Broker 服务器。温馨提示：如果上面在安装过程中发生了错误，大家可以查看user_home为用户主目录。该目录下会有众多的日志文件，如果一开始对这些文件的含义不了解也没关系，大家可以通过 ls-l 命令，逐一查看文件大小不为的文件，从而寻找错误日志，便于快速解决问题。RocketMQ 提供了众多的运维命令来查看 RocketMQ 集群的运行状态，在这里我先简单使用 clusterList 命令来查看集群的状态，用于验证一下集群的状态。sh./mqadmin clusterList-n 127.0.0.1:9876本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.3 实战：RocketMQ 学习环境搭建指南篇1.3 实战：RocketMQ 学习环境搭建指南篇Step3：使用 maven 命令编译源代码。mvn cleanpackage-DskipTests编译后在 target 目录下会生成可运行的 jar 包，如下图所示：Step4：我们可以将该包复制到自己常用的软件安装目录，例如笔者喜欢将其放在/opt/application 下。cp rocketmq-console-ng-1.0.0.jar/opt/application/Step5：启动 rocketmq-conolsenohup java-jar rocketmq-console-ng-1.0.0.jar&本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.3 实战：RocketMQ 学习环境搭建指南篇1.3 实战：RocketMQ 学习环境搭建指南篇Step2：namesrv/src/main/java/org/apache/rocketmq/namesrv/NamesrvStartup 设置环境变量 ROCKETMQ_HOME，操作步骤如下图所示：设置环境变量名称：ROCKETMQ_HOME，其值用于指定 RocketMQ 运行的主目录，笔者设置的路径为：/home/dingwpmz/tmp/rocketmq。本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.3 实战：RocketMQ 学习环境搭建指南篇1.3 实战：RocketMQ 学习环境搭建指南篇vi broker.conf#使用如下配置文件brokerClusterName=DefaultClusterbrokerName=broker-abrokerId=0deleteWhen=04fileReservedTime=48brokerRole=ASYNC_MASTERflushDiskType=ASYNC_FLUSHstorePathRootDir=/home/dingwpmz/tmp/rocketmq/storestorePathCommitLog=/home/dingwpmz/tmp/rocketmq/store/commitlognamesrvAddr=127.0.0.1:9876brokerIP1=192.168.3.10brokerIP2=192.168.3.10autoCreateTopicEnable=trueStep7：broker/src/main/java/org/apache/rocketmq/broker/BrokerStartup 设置环境变量 ROCKETMQ_HOME，操作步骤如下图所示：Step8：以 Debug 模式运行 BrokerStartup，其运行结果如下图所示：看到这样的提示就表示大功告成。本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.3 实战：RocketMQ 学习环境搭建指南篇1.4 RocketMQ HA 核心工作机制1.4 RocketMQ HA 核心工作机制温馨提示：建议参考代码 RocketMQ4.4 版本，4.5 版本引入了多副本机制，实现了主从自动切换，本文并不关心主从切换功能。一一、初识主从同步、初识主从同步主从同步基本实现过程如下图所示：RocketMQ 的主从同步机制如下：首先启动 Master 并在指定端口监听；客户端启动，主动连接 Master，建立 TCP 连接；客户端以每隔 5s 的间隔时间向服务端拉取消息，如果是第一次拉取的话，先获取本地commitlog 文件中最大的偏移量，以该偏移量向服务端拉取消息；服务端解析请求，并返回一批数据给客户端；本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.4 RocketMQ HA 核心工作机制1.4 RocketMQ HA 核心工作机制axRatio()/100.0);/2getResult.setSuggestPullingFromSlave(diff memory);/3代码1：首先介绍一下几个局部变量的含义：maxOffsetPy当前最大的物理偏移量。返回的偏移量为已存入到操作系统的 PageCache 中的内容。maxPhyOffsetPulling本次消息拉取最大物理偏移量，按照消息顺序拉取的基本原则，可以基本预测下次开始拉取的物理偏移量将大于该值，并且就在其附近。diffmaxOffsetPy 与 maxPhyOffsetPulling 之间的间隔，getMessage 通常用于消息消费时，即这个间隔可以理解为目前未处理的消息总大小。代码2：获取 RocketMQ 消息存储在 PageCache 中的总大小，如果当RocketMQ 容量超过该阔值，将会将被置换出内存，如果要访问不在 PageCache 中的消息，则需要从磁盘读取。StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE返回当前系统的总物理内存。参数accessMessageInMemoryMaxRatio设置消息存储在内存中的阀值，默认为 40。结合代码2这两个参数的含义，算出RocketMQ消息能映射到内存中最大值为40%*(机器物理内存)。代码3：设置下次拉起是否从从拉取标记，触发下次从从服务器拉取的条件为：当前所有可用消息数据(所有 commitlog)文件的大小已经超过了其阔值，默认为物理内存的40%。那 GetResult 的 suggestPullingFromSlave 属性在哪里使用呢？本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.4 RocketMQ HA 核心工作机制1.4 RocketMQ HA 核心工作机制代码2：如果当前服务器的角色为从服务器:并且 slaveReadEnable=true，则忽略代码1 设置的值，下次拉取切换为从主拉取。代码3：如果 slaveReadEnable=true(从允许读)，并且建议从从服务器读取，则从消息消费组建议当消息消费缓慢时建议的拉取 brokerId，由订阅组配置属性 whichBrokerWhenConsumeSlowly 决定；如果消息消费速度正常，则使用订阅组建议的 brokerId拉取消息进行消费，默认为主服务器。如果不允许从可读，则固定使用从主拉取。温馨提示：请注意 broker 服务参数 slaveReadEnable，与订阅组配置信息：whichBrokerWhenConsumeSlowly、brokerId 的值，在生产环境中，可以通过 updateSubGroup 命令动态改变订阅组的配置信息。如果订阅组的配置保持默认值的话，拉取消息请求发送到从服务器后，下一次消息拉取，无论是否开启 slaveReadEnable，下一次拉取，还是会发往主服务器。上面的步骤，在消息拉取命令的返回字段中，会将下次建议拉取 Broker 返回给客户端，根据其值从指定的 broker 拉取。消息拉取实现PullAPIWrapper在处理拉取结果时会将服务端建议的brokerId更新到broker 拉取缓存表中。在发起拉取请求之前，首先根据如下代码，选择待拉取消息的 Broker。本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.4 RocketMQ HA 核心工作机制1.4 RocketMQ HA 核心工作机制既然集群模式下消息消费进度存储在 Broker 端，当主服务器正常时，消息消费进度文件存储在主服务器，那提出如下两个问题：1.消息消费端在主服务器存活的情况下，会优先向主服务器反馈消息消费进度，那从服务器是如何同步消息消费进度的。2.当主服务器宕机后则消息消费端会向从服务器反馈消息消费进度，此时消息消费进度如何存储，当主服务器恢复正常后，主服务器如何得知最新的消息消费进度。为了解开上述两个疑问，我们优先来看一下 Broker 服务器在收到提交消息消费进度反馈命令后的处理逻辑：客户端定时向 Broker 端发送更新消息消费进度的请求，其入口为：RemoteBrokerOffsetStore#updateConsumeOffsetToBroker，该方法中一个非常关键的点是：选择broker 的逻辑，如下所示：如果主服务器存活，则选择主服务器，如果主服务器宕机，则选择从服务器。也就是说，不管消息是从主服务器拉取的还是从从服务器拉取的，提交消息消费进度请求，优先选择主服务器。服务端就是接收其偏移量，更新到服务端的内存中，然后定时持久化到$ROCKETMQ_HOME/store/config/consumerOffset.json。本文来自中间件兴趣圈公众号，仅作技术交流，未授权任何商业行为。1.4 RocketMQ HA 核心工作机制1.4 RocketMQ HA 核心工作机制第二种是，消息消费者在向主服务器拉取消息时，如果是是主服务器，在处理消息拉取时，也会更新消息消费进度。主服务器消息拉取时更新消息消费进度主服务器消息拉取时更新消息消费进度主服务器在处理消息拉取命令时，会触发消息消费进度的更新，其代码入口为：PullMessageProcessor#processRequestboolean storeOffsetEnable=brokerAllowSuspend;/1storeOffsetEnable=storeOffsetEnable&hasCommitOffsetFlag;storeOffsetEnable=storeOffsetEnable&this.brokerController.getMessageStoreConfig().getBrokerRole()!=BrokerRole.SLAVE;/2if(storeOffsetEnable)this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel),requestHeader.getConsumerGroup(),requestHeader.getTopic(),req