核心组件（4个组件+消息存储结构）

客户端消费模式

1. MQ的使用场景

昨天在写完之后，有些读者在评论中提出：到底什么时候用MQ？举几个典型的例子。

1.最常用的就是Publish/Subscribe

从开发模式中，我们使用过JAVA Swing Button EventListener或者使用JQuery 的自定义事件，由某个动作触发然后传播下去，这个事件广播就是Publish，我们监听的过程，就是订阅的过程。这个是MQ最基本的功能。

举个业务例子：订单支付的过程，会牵扯到会员模块、消息短信推送服务、订单更新，最初我们还能想清楚，所有把代码写在一个service中，但后续逐渐加业务，例如：会员还需要有积分、会员余额要预警、会员等级也得变化、还得告诉商户费用结清了。我举的例子可能不是很恰当，但这确实反映问题了，系统过于耦合。从这里引入MQ，对周边业务做清理，附属业务直接订阅消息。

2.应对大流量冲击、削峰填谷

系统的前端数据采集设备数量太大，且具有业务波峰。还有就是典型的双11，促销活动会期间，系统会承受比正常流量高很多倍的冲击。此时，采集的数据不要求实时，购物订单也不需要实时反馈给用户，面对这种应用场景，使用消息队列可以实现消息的异步处理、请求的流量削峰作用。

3.异构系统的通信

直接上例子：一个系统有多个开发小组，由于功能特点分别使用了C++、JAVA、Go、python等多种语言实现，系统之间需要通信，使用事件驱动架构，那么这里可以引入MQ了，作为异构系统和事件驱动架构的backbone。

2. RocketMQ初识

首先应该看看RocketMQ的收发消息模型：

消息收发模型

上图，你能看出来：

生产者生产消息，可以放到队列中，多个消费者可以消费。

生产者的同一种消息，可以放到不同的队列中，由消费者消息。

实际部署图：

物理部署图

如上图所示， RocketMQ的部署结构有4部分组成：NameServer、Broker、Producer、Consumer。

这里我们抽象以下：分为客户端（Producer、Consumer）、服务端（Broker、NameServer）两部分。简单来说，就是客户端的收发消息、服务器接收消息并持久化。

2.1 客户端的功能有什么？

client发送消息有负载均衡，因为客户端中保存着当前所有服务器列表，每次发送都切换一台服务器发送消息，服务均衡负载。

发送代码为线程安全，支持高并发写操作。

消费者端负载均衡集群消费模式下，同一个ID的所有消费者实例平均消费该Topic的所有队列。这里要注意，广播模式下，则一个consumer实例消费这个Topic对应的所有队列。这点和Apache ActiveMQ的主题订阅是一样的，每个消费者消费Topic的所有内容（若存在消费者消费慢，容易内存溢出）。

2.2 服务端（Broker）的功能有什么？

Broker就是RocketMQ的核心了，RocketMQ高并发读写主要利用Linux操作系统的PageCache特性，通过Java的MappedByteBuffer直接操作PageCache。MappedByteBuffer能直接将文件直接映射到内存，其实就是Map把文件的内容被映像到计算机虚拟内存的一块区域，这样就可以直接操作内存当中的数据而无需操作的时候每次都通过I/O去物理硬盘写文件的。

鼓励下自己，下面开始深入了

2.3 服务端存储的消息结构

RocketMQ高性能读写，得益于它的消息存储结构：commitLog和comsume queue两部分组成。（这部分大概了解，比别人多掌握一点）

数据存储结构

commitLog

保存所有消息的元数据，所有消息到达Broker之后都会保存到commitLog中，所有的topic消息都会写入一个commitLog中，通常的目录是：

${user.home}/store/${commitlog}/${fileName}

每个commitLog上限是1G，满了之后会自动新建一个commitLog文件保存数据。这里的Log文件会有清理机制，有两种清理措施：

按时间清理，默认清理3天前的commitLog。

按磁盘占比，当磁盘使用到75%，开始清理最老的commmitLog文件。

consumeQueue

ConsumeQueue相当于CommitLog的索引文件，消费者消费时会从consumeQueue中查找消息在commitLog中的offset，再去commitLog中查找元数据。如果某个消息只在CommitLog中有数据，没在ConsumerQueue中， 则消费者无法消费

consumeQueue的数据结构包含3个部分： consumeQueue中存储单元是一个20字节定长的二进制数据，顺序写顺序读 。如图：

ConsumeQueue存储格式的特性，保证了写过程的顺序写盘（写CommitLog文件），在单台服务器上，MQ元数据都落在单个文件上，大量数据IO都在顺序写同一个commitLog，满1G了再写新的，真正意义上的顺序写盘，再加上MQ默认是累计4K才强制从PageCache中刷到磁盘（缓存），所以高并发写性能突出，至于读取数据呢，有Queue的offset、size，让读盘操作是跳跃式的。在RocketMQ中读取消息是依赖系统PageCache，PageCache命中率越高，读性能越高，Linux平时也会尽量预读数据，使得应用直接访问磁盘的概率降低。

3. 核心组件

3.1 Name Server

用于存储Topic、Broker的关系，简单、稳定。多个Name-Server之间不通信，单台NameServer宕机不影响其他NameServer与集群；及时整个NameServer集群宕机，已经正常工作的Producer、Consumer、Broker仍然正常工作，但新加入的就无法工作。

NameServer压力不会大，主要用户维持心跳和提供Toptic-Broker的关系。但如过topic过多，导致Nameserver心跳数据过大，网络不好的情况下，传输失败，导致NameServer误认为Broker心跳失败。

3.2 Broker

Broker部署相对复杂：

Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master。

Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义。（BrokerId为0表示Master，非0表示Slave）

Master也可以部署多个，每个Broker与Name Server集群中的所有节点建立长连接，定时注册Topic信息到所有Name Server。

具有高并发读写，负载均衡和动态伸缩的特性：

负载均衡：Broker上存Topic信息，Topic由多个队列组成，队列会平均分散在多个Broker上，而Producer的发送机制保证消息尽量平均分布到所有队列中，最终效果就是所有消息都平均落在每个Broker上。

动态伸缩：Topic维度：假如一个Topic的消息量特别大，但集群水位压力还是很低，就可以扩大该Topic的队列数，Topic的队列数跟发送、消费速度成正比。Broker维度：如果集群水位很高了，需要扩容，直接加机器部署Broker就可以。Broker起来后想Namesrv注册，Producer、Consumer通过Namesrv发现新Broker，立即跟该Broker直连，收发消息。

高可用与可靠性

高可用：Broker提供主备，主宕机，备提供读，不可写。

可靠性：所有发送到Broker的消息，有同步刷盘和异步刷盘。同步刷盘时，消息写入物理文件才会返回成功。异步刷盘时，只有机器宕机，才会产生消息丢失，broker挂掉可能会发生，但是机器宕机崩溃是很少发生的，除非突然断电。

Broker与NameServer的心跳 ，单个Broker跟所有Namesrv保持心跳请求，心跳间隔为30秒，心跳请求中包括当前Broker所有的Topic信息。Namesrv会反查Broer的心跳信息，如果某个Broker在2分钟之内都没有心跳，则认为该Broker下线，调整Topic跟Broker的对应关系。但此时Namesrv不会主动通知Producer、Consumer有Broker宕机。

3.3 Producer

Producer启动时，也需要指定Namesrv的地址，从Namesrv集群中随机选一台建立长连接。如果该Namesrver宕机，会自动连其他Nameserver。直到有可用的Namesrv为止。

Producer每30秒从Namesrv获取Topic跟Broker的映射关系，更新到本地内存中。Producer会和它要发送的topic相关的master类型的broker建立TCP连接，每隔30秒发一次心跳。在Broker端也会每10秒扫描一次当前注册的Producer，如果发现某个Producer超过2分钟都没有发心跳，则断开连接。

生产者发送时，会自动轮询当前所有可发送的broker，一条消息发送成功，下次换另外一个broker发送，以达到消息平均落到所有的broker上。

3.4 Consumer

消费者启动时需要指定Namesrv地址，随机与其中一个Namesrv建立长连接。消费者每隔30秒从nameserver获取所有topic的最新队列情况（这意味着某个broker如果宕机，客户端最多要30秒才能感知），并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

Consumer跟Broker是长连接，会每隔30秒发心跳信息到Broker。Broker端每10秒检查一次当前存活的Consumer，若发现某个Consumer 2分钟内没有心跳，就断开与该Consumer的连接，并且向该消费组的其他实例发送通知，触发该消费者集群的负载均衡。

3.5 通信关系

Producer和Name Server：每一个Producer会与Name Server集群中的一台机器建立TCP连接，会从这台Name Server上拉取路由信息。

Producer和broker：Producer会和它要发送的topic相关的master类型的broker建立TCP连接，用于发送消息以及定时的心跳信息。broker中会记录该Producer的信息，供查询使用

broker与Name Server：broker（不管是master还是slave）会和每一台Name Server机器来建立TCP连接。broker在启动的时候会注册自己配置的topic信息到Name Server集群的每一台机器中。即每一台Name Server都有该broker的topic的配置信息。master与master之间无连接，master与slave之间有连接

Consumer和Name Server：每一个Consumer会和Name Server集群中的一台机器建立TCP连接，会从这台Name Server上拉取路由信息，进行负载均衡

Consumer和broker：Consumer可以与master或者slave的broker建立TCP连接来进行消费消息，Consumer也会向它所消费的broker发送心跳信息，供broker记录。

4. 消费模式

4.1 消费者端的消费以及负载均衡

先讨论消费者的消费模式，消费者有两种模式消费：

集群消费

广播消费。

广播消费

每个消费者消费Topic下的所有队列。

集群消费

一个topic可以由同一个ID下所有消费者分担消费。具体例子：假如TopicA有6个队列（kafka称之为分区），某个消费者ID起了2个消费者实例，那么每个消费者负责消费3个队列。如果再增加一个消费者ID相同消费者实例，即当前共有3个消费者同时消费6个队列，那每个消费者负责2个队列的消费。

消费者端的负载均衡，就是集群消费模式下，同一个ID的所有消费者实例平均消费该Topic的所有队列。

对于负载均衡，在出现分区（kafka称为分区）或者队列（RocketMQ称队列）增加或者减少的时候、Consumer增加或者减少的时候都会进行reblance操作。

对于RocketMQ：客户端自己会定时对所有的topic的进行reblance操作，对于每个topic，会从broker获取所有Consumer列表，从broker获取队列列表，按照负载均衡策略，计算各自负责哪些队列。这种就要求进行负载均衡的时候，各个Consumer获取的数据是一致的，不然不同的Consumer的reblance结果就不同。

遍历Consumer下的所有topic，然后根据topic订阅所有的消息

获取同一topic和Consumer Group下的所有Consumer

然后根据具体的分配策略来分配消费队列，分配的策略包含：平均分配、消费端配置等

4.2 客户端消费是broker的Push还是客户端的Pull？

实际上consumer的消费只有主动拉的方式，那么有没有push的方式？实际上类似http的请求长连接，consumer发起pull请求，等待broker有数据之后，再把数据push回来，实际上还是主动拉消息。

以上我们对Rocket做了全面的理论探索，后续配合实施外加代码实战，带你玩转RocketMQ。