读懂消息队列：Kafka与RocketMQ

曾经终败给现在 2023-07-19 10:56 18阅读 0赞

3月份学完了极客时间的《消息列队高手课》专栏，专栏讲解了许多消息队列的基础知识并且对Kafka与RocketMQ两种主流消息队列有精彩的对比分析。学完专栏后将所有要点整理为笔记记录下来，其他相关知识也搜索了大量资料，博文写得比较凌乱，分为两部分，第一部分是消息队列的基础知识，不涉及具体的消息队列产品。在了解了基础知识后，第二部分着重比较两款消息队列的明星产品——RocketMQ与Kafka，在比较的过程中理解消息队列产品的设计与架构。

# 消息队列基础知识 #

## 1、如何确保消息可靠传递 ##

我们一先来看一下，整个消息从生产到消费的过程中，哪些地方可能会导致丢消息，以及应该如何避免消息丢失。你可以看下这个图，一条消息从生产到消费完成这个过程，可以划分三个阶段，为了方便描述，我给每个阶段分别起了个名字。

![20200406114518705.png][]

### 生产阶段 ###

消息队列通过最常用的请求确认机制，来保证消息的可靠传递：当你的代码调用发消息方法时，消息队列的客户端会把消息发送到Broker，Broker收到消息后，会给客户端返回一个确认响应，表明消息已经收到了。客户端收到响应后，完成了一次正常消息的发送。只要Producer收到了Broker的确认响应，就可以保证消息在生产阶段不会丢失。有些消息队列在长时间没收到发送确认响应后，会自动重试，如果重试再失败，就会以返回值或者异常的方式告知用户。

在编写发送消息代码时，需要注意，正确处理返回值或者捕获异常，就可以保证这个阶段的消息不会丢失。以 Kafka 为例，我们看一下如何可靠地发送消息。

同步发送时，只要注意捕获异常即可。

try { 
        RecordMetadata metadata = producer.send(record).get();
        System.out.println("消息发送成功。");
    } catch (Throwable e) { 
        System.out.println("消息发送失败！");
        System.out.println(e);
    }

异步发送时，则需要在回调方法里进行检查。这个地方是需要特别注意的，很多丢消息的原因就是，我们使用了异步发送，却没有在回调中检查发送结果。

producer.send(record, (metadata, exception) -> { 
        if (metadata != null) { 
            System.out.println("消息发送成功。");
        } else { 
            System.out.println("消息发送失败！");
            System.out.println(exception);
        }
    });

另外这里推荐为Producer的retries （重试次数）设置一个比较合理的值，一般是3 ，但是为了保证消息不丢失的话一般会设置比较大一点。设置完成之后，当出现网络问题之后能够自动重试消息发送，避免消息丢失。另外，建议还要设置重试间隔，因为间隔太小的话重试的效果就不明显了，网络波动一次你3次一下子就重试完了。

### 存储阶段 ###

在存储阶段正常情况下，只要Broker在正常运行，就不会出现丢失消息的问题，但是如果Broker出现了故障，比如进程死掉了或者服务器宕机了，还是可能会丢失消息的。如果对消息的可靠性要求非常高，可以通过配置Broker参数来避免因为宕机丢消息。

对于单个节点的Broker，需要配置Broker参数，在收到消息后，将消息写入磁盘后再给Producer返回确认响应，这样即使发生宕机，由于消息已经被写入磁盘，就不会丢失消息，恢复后还可以继续消费。

例如，在RocketMQ中，需要将刷盘方式flushDiskType配置为SYNC\_FLUSH同步刷盘。\*\*如果Broker是由多个节点组成的集群，需要将Broker集群配置成：至少将消息发送到2个以上的节点，再给客户端回复发送确认响应。这样当某个Broker宕机时，其他的Broker可以替代宕机的Broker，也不会发生消息丢失。

Kafka为分区（Partition）引入了多副本（Replica）机制。分区（Partition）中的多个副本之间会有一个叫做leader的家伙，其他副本称为 follower。我们发送的消息会被发送到leader，然后follower才能从leader中拉取消息进行同步。生产者和消费者只与leader交互。你可以理解为其他副本只是leader的拷贝，它们的存在只是为了保证消息存储的安全性。

试想一种情况：假如leader所在的broker突然挂掉，那么就要从follower副本重新选出一个leader，但是leader的数据如果还有一些没有被follower副本的同步的话，就会造成消息丢失。解决办法就是我们设置acks=all。acks是Kafka生产者(Producer) 很重要的一个参数。acks的默认值即为1，代表我们的消息被leader副本接收之后就算被成功发送。当我们配置acks=all代表所有副本都接收到该消息之后该消息才算真正成功被发送。

设置replication.factor >= 3，为了保证leader能有多个follower副本，我们一般会为topic设置replication.factor >= 3。这样就可以保证每个分区(partition) 至少有3个副本。虽然造成了数据冗余，但是带来了数据的安全性。

设置min.insync.replicas > 1，这样配置代表消息至少要被写入到2个副本才算是被成功发送。min.insync.replicas的默认值为1，在实际生产中应尽量避免默认值1。

### 消费阶段 ###

对于Kafka，消息在被追加到Partition（分区）的时候都会分配一个特定的偏移量（offset）。偏移量（offset）表示Consumer当前消费到的 Partition（分区）的所在的位置。Kafka通过偏移量（offset）可以保证消息在分区内的顺序性。

消费阶段采用和生产阶段类似的确认机制来保证消息的可靠传递，客户端从Broker拉取消息后，执行用户的消费业务逻辑，成功后，才会给Broker发送消费确认响应。如果Broker没有收到消费确认响应，下次拉消息的时候还会返回同一条消息，确保消息不会在网络传输过程中丢失，也不会因为客户端在执行消费逻辑中出错导致丢失。

在编写消费代码时需要注意的是，不要在收到消息后就立即发送消费确认，而是应该在执行完所有消费业务逻辑之后，再发送消费确认。对于Kafka应手动关闭消费者拉取消息后自动提交offset的功能，每次在真正消费完消息之后之后再由消费者手动提交offset。

## 2、如何处理消息重复 ##

首先需要明确的一点是：**消息重复的情况必然存在**。

在MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议）中，给出了三种传递消息时能够提供的服务质量标准，这三种服务质量从低到高依次是：

*  At most once: 至多一次。消息在传递时，最多会被送达一次。换一个说法就是，没什么消息可靠性保证，允许丢消息。一般都是一些对消息可靠性要求不太高的监控场景使用。
 *  At least once: 至少一次。消息在传递时，至少会被送达一次。也就是说，不允许丢消息，但是允许有少量重复消息出现。
 *  Exactly once：恰好一次。消息在传递时，只会被送达一次，不允许丢失也不允许重复，这个是最高的等级。

这个服务质量标准不仅适用于MQTT，对所有的消息队列都是适用的。我们现在常用的绝大部分消息队列提供的服务质量都是At least once，包括RocketMQ、RabbitMQ和Kafka都是这样。

既然消息队列无法保证消息不重复，就需要我们的**消费代码能够接受“消息是可能会重复的”这一现状，然后，通过一些方法来消除重复消息对业务的影响。一般解决重复消息的办法是，在消费端，让我们消费消息的操作具备幂等性**。

At least once + 幂等消费 = Exactly once，以下是几种常用的设计幂等操作的方法。

1.  数据库唯一约束实现幂等，以消息体中某个具备唯一特性的字段为唯一索引建立消息流水表，每消费一条消息就插入一行记录。
2.  乐观锁实现幂等，消息体中传递数据的version，获取数据时根据version来查，消费成功后更新version
3.  分布式锁幂等

## 3、消息模型：主题与队列 ##

最初的消息队列，就是一个严格意义上的队列。在计算机领域，“队列（Queue）”是一种数据结构，有完整而严格的定义。

早期的消息队列，就是按照“队列”的数据结构来设计的。我们一起看下这个图，生产者（Producer）发消息就是入队操作，消费者（Consumer）收消息就是出队也就是删除操作，服务端存放消息的容器自然就称为“队列”。

![20200406114248313.png][]

如果有多个生产者往同一个队列里面发送消息，这个队列中可以消费到的消息，就是这些生产者生产的所有消息的合集。消息的顺序就是这些生产者发送消息的自然顺序。如果有多个消费者接收同一个队列的消息，这些消费者之间实际上是竞争的关系，每个消费者只能收到队列中的一部分消息，也就是说任何一条消息只能被其中的一个消费者收到。如果需要将一份消息数据分发给多个消费者，要求每个消费者都能收到全量的消息，例如，对于一份订单数据，风控系统、分析系统、支付系统等都需要接收消息。这个时候，单个队列就满足不了需求，一个可行的解决方式是，为每个消费者创建一个单独的队列，让生产者发送多份。显然这是个比较蠢的做法，同样的一份消息数据被复制到多个队列中会浪费资源，更重要的是，生产者必须知道有多少个消费者。为每个消费者单独发送一份消息，这实际上违背了消息队列“解耦”这个设计初衷。

![20200406114307773.jpg][]

在发布—订阅模型中，消息的发送方称为发布者（Publisher），消息的接收方称为订阅者（Subscriber），服务端存放消息的容器称为主题（Topic）。发布者将消息发送到主题中，订阅者在接收消息之前需要先“订阅主题”。“订阅”在这里既是一个动作，同时还可以认为是主题在消费时的一个逻辑副本，每份订阅中，订阅者都可以接收到主题的所有消息。在消息领域的历史上很长的一段时间，队列模式和发布—订阅模式是并存的，有些消息队列同时支持这两种消息模型，比如ActiveMQ。我们仔细对比一下这两种模型，生产者就是发布者，消费者就是订阅者，队列就是主题，并没有本质的区别。它们最大的区别其实就是，一份消息数据能不能被消费多次的问题。实际上，在这种发布—订阅模型中，如果只有一个订阅者，那它和队列模型就基本是一样的了。也就是说，发布—订阅模型在功能层面上是可以兼容队列模型的。

![20200406114407345.jpg][]

## 4、事务消息保障业务处理与消息投递的原子性 ##

事务消息需要消息队列提供相应的功能才能实现，Kafka和RocketMQ都提供了事务相关功能。以在电商平台上下单购物的场景为例。

![20200406121353719.jpg][]

首先，订单系统在消息队列上开启一个事务。然后订单系统给消息服务器发送一个“半消息”，这个**半消息不是说消息内容不完整，它包含的内容就是完整的消息内容，半消息和普通消息的唯一区别是，在事务提交之前，对于消费者来说，这个消息是不可见的**。

半消息发送成功后，订单系统就可以执行本地事务了，在订单库中创建一条订单记录，并提交订单库的数据库事务。然后根据本地事务的执行结果决定提交或者回滚事务消息。如果订单创建成功，那就提交事务消息，购物车系统就可以消费到这条消息继续后续的流程。如果订单创建失败，那就回滚事务消息，购物车系统就不会收到这条消息。这样就基本实现了最终一致性事务。

如果你足够细心，可能已经发现了，这个实现过程中，有一个问题是没有解决的。如果在第四步提交事务消息时失败了怎么办？对于这个问题，Kafka和RocketMQ给出了两种不同的解决方案。Kafka的解决方案比较简单粗暴，直接抛出异常，让用户自行处理。我们可以在业务代码中反复重试提交，直到提交成功，或者删除之前创建的订单进行补偿。RocketMQ则给出了另外一种解决方案，如下图所示。

![2020040612071785.jpg][]

在RocketMQ中的事务实现中，增加了事务反查的机制来解决事务消息提交失败的问题。如果Producer也就是订单系统，在提交或者回滚事务消息时发生网络异常，RocketMQ的Broker没有收到提交或者回滚的请求，Broker会定期去Producer上反查这个事务对应的本地事务的状态，然后根据反查结果决定提交或者回滚这个事务。

为了支撑这个事务反查机制，我们的业务代码需要实现一个反查本地事务状态的接口，告知RocketMQ本地事务是成功还是失败。

通过上面这幅图，我们可以看出，事务不一致的两种情况中，永远不会发生“订单创建失败，购物车处理成功（从购物车中移除商品）”这样的情况。因为如果A服务本地事务都失败了，那B服务永远不会执行任何操作，因为消息压根就不会传到B服务。

那么订单创建成功，购物车处理失败这种情况会不会发生呢。答案是会的，因为A服务只负责当我消息执行成功了，保证消息能够送达到B，至于B服务接到消息后最终的执行结果成不成功A并不管。那B服务失败了怎么办？如果B最终执行失败，几乎可以断定就是代码有问题所以才引起的异常，因为消费端RocketMQ有重试机制，如果不是代码问题一般重试几次就能成功。如果是代码的原因引起多次重试失败后，也没有关系，将该异常记录下来，由人工处理，人工兜底处理后，就可以让事务达到最终的一致性。

RocketMQ是一种最终一致性的分布式事务，就是说它保证的是消息最终一致性，而不是像2PC、3PC、TCC那样实现强一致的分布式事务。

## 5、顺序消费消息 ##

哪些场景要求顺序消费？

**最常见的场景就是数据同步（从一个库同步到另一个库），当数据量大的时候数据同步压力也是很大的。这种情况我们都是怼到队列里面去，然后慢慢消费的，那问题就来了呀，我们在数据库同时对一个Id的数据进行了增、改、删三个操作，但是你消息发过去消费的时候变成了改，删、增，这样数据就不对了**。

本来一条数据应该删掉了，结果在你那儿却还在，这不是出大问题！

![20200827151859596.png][]

通常我们所说的顺序消费消息指的是**生产者按照顺序发送，消费者按照顺序进行消费**，听起来简单，但做起来却非常困难。**典型的要求消息顺序消费的场景就是基于binlog作数据同步**。

我们都知道无论是Kafka还是RocketMQ，每个主题下面都有若干分区（RocketMQ叫队列），如果消息被分配到不同的分区中，那么Kafka是不能保证消息的消费顺序的，因为每个分区都分配到一个消费者，此时无法保证消费者的消费先后，**因此如果需要进行消息具有消费顺序性，可以在生产端指定这一类消息的key，这类消息都用相同的key进行消息发送，kafka就会根据key哈希取模选取其中一个分区进行存储，由于一个分区只能由一个消费者进行监听消费，因此这时候消息就具有消息消费的顺序性了**。

### 发送端 ###

但以上情况只是在正常情况下可以保证顺序消息，但发生故障后，就没办法保证消息的顺序了，我总结以下两点：

*  当生产端是异步发送时，此时有消息发送失败，比如你异步发送了1、2、3消息，2消息发送异常要重发，这时候顺序就乱了；
 *  当Broker宕机出现问题，此时生产端有可能会把顺序消息发送到不同的分区，这时会发生短暂消息顺序不一致的现象。

针对以上两点，**生产端必须保证单线程同步发送，这还好解决，针对第二点，想要做到严格的消息顺序，就要保证当集群出现故障后集群立马不可用，或者主题做成单分区，但这么做大大牺牲了集群的高可用，单分区也会另集群性能大大降低**。

### 存储端 ###

对于存储端，要保证消息顺序，会有以下几个问题：

*  消息不能分区，也就是1个topic，只能有1个队列。在Kafka中，它叫做partition；在RocketMQ中，它叫做queue。 如果你有多个队列，那同1个topic的消息，会分散到多个分区里面，自然不能保证顺序。
 *  即使只有1个队列的情况下，会有第2个问题。Broker挂了之后，能否切换到其他机器？也就是高可用问题。

比如你当前的broker挂了，上面还有消息没有消费完。此时切换到其他机器，可用性保证了，但消息顺序就乱掉了。

要想保证，一方面要同步复制，不能异步复制；另1方面得保证，切机器之前，挂掉的机器上面，所有消息必须消费完了，不能有残留。很明显，这个很难！！！

### 消费端 ###

对于**消费端，不能并行消费，即不能开多线程或者多个客户端消费同1个队列**。RocketMQ会为每个队列分配一个PullRequest，并将其放入pullRequestQueue，PullMessageService线程会不断轮询从pullRequestQueue中取出PullRequest去拉取消息，接着将拉取到的消息给到ConsumeMessageService处理，ConsumeMessageService有两个子接口：

// 并发消息消费逻辑实现类
    org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService;
    // 顺序消息消费逻辑实现类
    org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService;

ConsumeMessageConcurrentlyService内部有一个线程池，用于并发消费，同样地，如果需要顺序消费，那么RocketMQ提供了 ConsumeMessageOrderlyService类进行顺序消息消费处理。从ConsumeMessageOrderlyService源码中能够看出RocketMQ能够实现局部消费顺序，主要有以下两点：

*  RocketMQ会为每个消息队列建一个对象锁，在消费某一个消息消费队列时先加锁，意味着一个消费者内消费线程池中的并发度是消费队列级别，同一个消费队列在同一时刻只会被一个线程消费，其他线程排队消费。保证了当前Consumer内，同一队列的消息进行串行消费。
 *  向Broker端请求锁定当前顺序消费的队列，防止在消费过程中被分配给其它消费者处理从而打乱消费顺序。

从上面的分析可以看出，要保证消息的严格有序，有多么困难！**发送端和接收端的问题，还好解决一点，限制异步发送，限制并行消费。但对于存储端，机器挂了之后，切换的问题，就很难解决了。你切换了，可能消息就会乱；你不切换，那就暂时不可用。这2者之间，就需要权衡了。结论是：无论RocketMQ还是Kafka，都不保证消息的严格有序消费！**

# RocketMQ与Kafka对比 #

## 1、Kafka的架构及消息存储 ##

在所有的存储系统中，消息队列的存储可能是最简单的。每个主题包含若干个分区，每个分区其实就是一个WAL（WriteAheadLog），写入的时候只能在尾部追加，不允许修改。读取的时候，根据一个索引序号进行查询，然后连续顺序往下读。

如下图，一个Kafka架构包括若干个Producer（服务器日志、业务数据、web前端产生的page view等），若干个Broker（Kafka支持水平扩展，一般broker数量越多集群的吞吐量越大），若干个consumer group，一个Zookeeper集群，Kafka通过Zookeeper管理集群配置、选举leader、consumer group发生变化时进行rebalance。  
![20200406130317797.png][]

**Topic & Partition的关系是什么**

一个topic为一类消息，每条消息必须指定一个topic。物理上，一个topic分成一个或多个partition，每个partition有多个副本分布在不同的broker中，如下图所示，一个机器可能既是topicA\_partition\_1的leader又是topicB\_partition\_2的follower。

![20200406123900110.png][]

每个partition在存储层面是一个append log文件，发布到此partition的消息会追加到log文件的尾部，为顺序写人磁盘（顺序写磁盘比随机写内存的效率还要高）。每条消息在log文件中的位置成为offset（偏移量），offset为一个long型数字，唯一标记一条消息。写入过程如下图所示，Kafka中只能保证partition中记录是有序的，而不保证topic中不同partition的顺序。

![20200406124049550.png][]

这种存储方式，对于每个文件来说是顺序IO，但是当并发的读写多个partition的时候，就对应多个文件的顺序IO，只要partition的数量足够大，表现在文件系统的磁盘层面还是随机IO。因此当出现了多人partition或者topic个数过多时，Kafka的性能会急剧下降。

## 2、RocketMQ的架构及消息存储 ##

RocketMQ的部署架构如下图所示，在早期的RocketMQ版本中，是有依赖ZK的。而现在的版本中已经去掉了对ZK的依赖，转而使用自己开发的NameServer来实现元数据（Topic路由信息）的管理，并且这个NameServer是无状态的，可以随意的部署多台，其代码也非常简单，非常轻量。

RocketMQ的启动流程可以描述为：

*  Broker消息服务器启动，向所有NameServer注册，NameServer与每台Broker服务器保持长连接，并定时检测 Broker是否存活，如果检测到broker宕机，则从路由注册表中将其移除。
 *  消息生产者(Producer)在发送消息之前先从NameServer获取Broker服务器地址列表，然后根据负载算法从列表中选择一台消息服务器进行消息发送。
 *  消息消费者（Consumer）在拉取消息之前先从NameServer获取Broker服务器地址列表，然后根据负载算法从列表中选择一台消息服务器进行消息拉取。

![20200330180807836.png][]

不同于Kafka里面，一台机器可以同时是Master和Slave。在RocketMQ里面，1台机器只能要么是Master，要么是Slave。这个在初始的机器配置里面，就定死了，其架构拓扑图如下。  
![20200406125607569.png][]

在这里，RocketMQ里面queue这个概念，就对应Kafka里面partition。图中有3个Master， 6个Slave，那对应到物理上面，就是9台机器3个broker。

通过对比可以看出，Kafka和RocketMQ在Master/Slave/Broker这个3个概念上的差异。这个差异，也就影响到topic&partition这种逻辑概念和Master/Slave/Broker这些物理概念上的映射关系。

具体来讲就是：在Kafka里面，Maser/Slave是选举出来的，而RocketMQ不需要选举！在Kafka里面，每个partition的Master是谁Slave是谁要通过选举决定。Master/Slave是动态的，当Master挂了之后，会有1个Slave切换成Master。

而在RocketMQ中，不需要选举，Master/Slave的角色也是固定的。当一个Master挂了之后，你可以写到其他Master上，但不会说一个Slave切换成Master。这种简化，使得RocketMQ可以不依赖ZooKeeper就很好的管理Topic&queue和物理机器的映射关系了，也实现了高可用。

为了解决Kafka的设计中当topic或partition过多，顺序IO变随机IO的问题，RocketMQ采用了单一的日志文件，即把同1台机器上面所有topic的所有queue的消息，存放在一个文件里面，从而避免了随机的磁盘写入。其存储结构如下图所示。

![20200406125023743.png][]

所有消息都存在一个单一的CommitLog文件里面（完全的顺序写），然后有后台线程异步的同步到ConsumeQueue，再由Consumer进行消费。

需要说明的是：Kafka针对Producer和Consumer使用了同1份存储结构，而RocketMQ却为Producer和Consumer分别设计了不同的存储结构，Producer对应CommitLog，Consumer对应ConsumeQueue。ConsumeQueue中并不需要存储消息的内容，而存储的是消息在CommitLog中的offset。也就是说，ConsumeQueue其实是CommitLog的一个索引文件。

这里之所以可以用“异步线程”，也是因为消息队列天生就是用来“缓冲消息”的。只要消息到了CommitLog，发送的消息也就不会丢。只要消息不丢，那就有了充足的回旋余地，用一个后台线程慢慢同步到ConsumeQueue，再由Consumer消费。可以说，这也是在消息队列内部的一个典型的“最终一致性”的案例：Producer发了消息，进了CommitLog，此时Consumer并不可见。但没关系，只要消息不丢，消息的offset最终肯定会写入ConsumeQueue，让Consumer可以消费。很显然，Consumer消费消息的时候，要读2次：先读ConsumeQueue得到offset，再读CommitLog得到消息内容（随机读）。

## 3、Kafka与RocketMQ的消息查找 ##

Kafka的存储以Partition为单位，每个Partition包含一组消息文件（Segment file）和一组索引文件（Index），并且消息文件和索引文件一一对应，具有相同的文件名（但文件扩展名不一样），文件名就是这个文件中第一条消息的索引序号。

每个索引中保存索引序号（也就是这条消息是这个分区中的第几条消息）和对应的消息在消息文件中的绝对位置。在**索引的设计上，Kafka采用的是稀疏索引，为了节省存储空间，它不会为每一条消息都创建索引，而是每隔几条消息创建一条索引**。

写入消息的时候非常简单，就是在消息文件尾部连续追加写入，一个文件写满了再写下一个文件。查找消息时，首先根据文件名找到所在的索引文件，然后用二分法遍历索引文件内的索引，在里面找到离目标消息最近的索引，再去消息文件中，找到这条最近的索引指向的消息位置，从这个位置开始顺序遍历消息文件，找到目标消息。

可以看到，寻址过程还是需要一定时间的。一旦找到消息位置后，就可以批量顺序读取，不必每条消息都要进行一次寻址。

RocketMQ的存储以Broker为单位。它的存储也是分为消息文件和索引文件，但是在RocketMQ中，每个Broker只有一组消息文件，它把在**这个 Broker上的所有主题的消息都存在这一组消息文件中**。索引文件和Kafka 一样，是按照主题和队列分别建立的，每个队列对应一组索引文件，这组索引文件在RocketMQ中称为ConsumerQueue。RocketMQ引入Hash索引机制**为消息建立定长稠密索引，它为每一条消息都建立索引，每个索引的长度（注意不是消息长度）是固定的20个字节**。

写入消息的时候，Broker上所有主题、所有队列的消息按照自然顺序追加写入到同一个消息文件中，一个文件写满了再写下一个文件。查找消息的时候，可以直接根据队列的消息序号，计算出索引的全局位置（索引序号 x 索引固定长度20），然后直接读取这条索引，再根据索引中记录的消息的全局位置，找到消息。可以看到，这里两次寻址都是绝对位置寻址，比Kafka的查找是要快的。

![20200416191721429.png][]

两种存储结构的对比如上图所示，可以看到它们有很多共通的地方，都是采用消息文件 + 索引文件的存储方式，索引文件的名字都是第1条消息的索引序号，索引中记录了消息的位置等等。

在消息文件的存储粒度上，Kafka以分区为单位，粒度更细，优点是更加灵活，很容易进行数据迁移和扩容。RocketMQ以Broker为单位，较粗的粒度牺牲了灵活性，带来的好处是，在写入的时候，同时写入的文件更少，有更好的批量写入性能（不同主题和分区的数据可以组成一批一起写入），更多的顺序写入，尤其是在Broker上有很多主题和分区的情况下，有更好的写入性能。

索引设计上，RocketMQ和Kafka分别采用了稠密和稀疏索引，稠密索引需要更多的存储空间，但查找性能更好，稀疏索引能节省一些存储空间，代价是牺牲了查找性能。

## 4、磁盘IO ##

RocketMQ和Kafka都基于磁盘做持久化，使用文件系统存储消息，两者都使用了PageCache，且都通过零拷贝（Zero Copy）的方式来提高IO读写性能，不同的是Kafka的零拷贝使用的是sendFile，而RocketMQ使用的是mmap虚拟内存映射的方式。

PageCache是OS对文件的缓存，用于加速对文件的读写。对于数据文件的写入，OS会先写入至Cache内，随后通过异步的方式由pdflush内核线程将Cache内的数据刷盘至物理磁盘上。对于数据文件的读取，如果一次读取文件时出现未命中PageCache的情况，OS从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取。对于文件的顺序读写操作来说，读和写的区域都在OS的PageCache内，此时读写性能接近于内存。

传统的IO读文件总结就是，一个操作中有2次上下文切换和2次数据copy（DMA Copy + CPU copy），传统的IO读写文件过程如下图：

![20200416195003896.png][]

1.  系统调用read导致了从用户空间到内核空间的上下文切换。DMA模块从磁盘中读取文件内容，并将其存储在内核空间的缓冲区内，完成了第1次复制。
2.  数据从内核空间缓冲区复制到用户空间缓冲区，之后系统调用read返回，这导致了从内核空间向用户空间的上下文切换。此时，需要的数据已存放在指定的用户空间缓冲区内。
3.  系统调用write导致从用户空间到内核空间的上下文切换。数据从用户空间缓冲区被再次复制到内核空间缓冲区，完成了第3次复制。不过，这次数据存放在内核空间中与使用的socket相关的特定缓冲区中，而不是步骤1中的缓冲区。
4.  系统调用返回，导致了第4次上下文切换。

上面的过程中存在很多的数据冗余。某些冗余可以被消除，以减少开销、提高性能。使用mmap或sendFile系统调用都可以减少1次copy。

OS的mmap内存映射技术，通过MMU（内存管理单元）映射文件，将文件直接映射到用户态的内存地址，使得对文件的操作不再是write/read，而转化为直接对内存地址的操作，使随机读写文件和读写内存相似的速度。mmap把文件映射到用户空间里的虚拟内存，省去了从内核缓冲区复制到用户空间的过程，文件中的位置在虚拟内存中有了对应的地址，可以像操作内存一样操作这个文件，这样的文件读写文件方式少了数据从内核缓存到用户空间的拷贝，效率很高，过程如下图所示：

![20200416202335460.png][]

1.  mmap系统调用导致文件的内容通过DMA模块被复制到内核缓冲区中，该缓冲区之后与用户进程共享，这样就内核缓冲区与用户缓冲区之间的复制就不会发生。
2.  write系统调用导致内核将数据从内核缓冲区复制到与socket相关联的内核缓冲区中。
3.  DMA模块将数据由socket的缓冲区传递给协议引擎时，第3次复制发生。

## 5、负载均衡 ##

Kafka：支持负载均衡。

1、一个broker通常就是一台服务器节点。对于同一个Topic的不同分区，Kafka会尽力将这些分区分布到不同的Broker服务器上，zookeeper保存了broker、topic和partition的元数据信息。分区首领会处理来自客户端的生产请求，kafka分区首领会被分配到不同的broker服务器上，让不同的broker服务器共同分担任务。每一个broker都缓存了元数据信息，客户端可以从任意一个broker获取元数据信息并缓存起来，根据元数据信息知道要往哪里发送请求。

2、kafka的消费者组订阅同一个topic，会尽可能地使得每一个消费者分配到相同数量的分区，分摊负载。

3、当消费者加入或者退出消费者组的时候，还会触发再均衡，为每一个消费者重新分配分区，分摊负载。kafka的负载均衡大部分是自动完成的，分区的创建也是kafka完成的，隐藏了很多细节，避免了繁琐的配置和人为疏忽造成的负载问题。

4、发送端由topic和key来决定消息发往哪个分区，如果key为null，那么会使用轮询算法将消息均衡地发送到同一个topic的不同分区中。如果key不为null，那么会根据key的hashcode取模计算出要发往的分区。

RocketMQ：支持负载均衡。

1、一个broker通常是一个服务器节点，broker分为master和slave，master和slave存储的数据一样，slave从master同步数据。nameserver与每个集群成员保持心跳，保存着Topic-Broker路由信息，同一个topic的队列会分布在不同的服务器上。

2、发送消息通过轮询队列的方式发送，每个队列接收平均的消息量。发送消息指定topic、tags、keys，无法指定投递到哪个队列（没有意义，集群消费和广播消费跟消息存放在哪个队列没有关系）。tags选填类似于Gmail为每封邮件设置的标签，方便服务器过滤使用。keys选填代表这条消息的业务关键词，服务器会根据keys创建哈希索引，设置后可以在Console系统根据Topic、Keys来查询消息，由于是哈希索引，要尽可能保证key唯一例如订单号、商品Id等。

3、RocketMQ的负载均衡策略规定：Consumer数量应该小于等于Queue数量，如果Consumer超过Queue数量，那么多余的Consumer将不能消费消息。这一点和kafka是一致的，RocketMQ会尽可能地为每一个Consumer分配相同数量的队列，分摊负载。

[20200406114518705.png]: /images/20230528/45c4e6c0178948af96a9c0b2d4b26c80.png
[20200406114248313.png]: /images/20230528/eaba462df172414084f5d28066629e11.png
[20200406114307773.jpg]: /images/20230528/379b799a16f848ffa679d4292203b537.png
[20200406114407345.jpg]: /images/20230528/5d145d90ac524abaa6427a2f83592224.png
[20200406121353719.jpg]: /images/20230528/fb2912bb31504712add76bf5147d5e83.png
[2020040612071785.jpg]: /images/20230528/df13a0b7f6e949db910483e7994dc47f.png
[20200827151859596.png]: /images/20230528/eee64fd50ea042288f42ceea5228c9aa.png
[20200406130317797.png]: /images/20230528/5dd03e67d8ef4a32a764f13781614df8.png
[20200406123900110.png]: /images/20230528/97f7ccf194c443d4ac07c083eaee8121.png
[20200406124049550.png]: /images/20230528/d85fe8ee9a834db48f058b307a51f4c4.png
[20200330180807836.png]: /images/20230528/57fa4f821c8b4e55849b177889e776ef.png
[20200406125607569.png]: /images/20230528/e235f37cba1c4911b576ee06ef70113e.png
[20200406125023743.png]: /images/20230528/e510bf6965ce417f9a21517a0d9f4304.png
[20200416191721429.png]: /images/20230528/076d54aff85b41658f5c9564b4d7a2d7.png
[20200416195003896.png]: /images/20230528/8c6e81dab76b4525a837d9ba0eebe07c.png
[20200416202335460.png]: /images/20230528/d24820cdc64744319f82f4c79f532658.png