前言

最近重新读起kafka的内容，看到kafka官网文档里，有专门一栏讲kafka的设计，觉得很受益。我们常常会知道这个中间件是什么，是什么机制，这次想换个角度来聊，它在设计消息系统的时候，都做了哪些考虑？为什么这么考虑？

kafka只有一个，但是它的设计思想正在被成百上千个组件在学习和借鉴。这或许是更有价值的一件事。

PS:如果有讲的不对的地方，欢迎大家指正。

PPS : 另外，本文比较长，偏基础，部分细节直接通过目录跳过也行 😂

那我开始了啊

消息中间件的编年史

在正式介绍kafka的相关细节特性之前，先插播一个消息中间件的编年史。

• 2001年提出JMS的消息模型和规范以来
• 2003年诞生了ActiveMQ，在今天它也还在不断迭代;
• 2007年的RabbitMQ，它是第一个基于应用层协议的开放标准AMQP的协议实现；
• 2011年kafka,也就是我们今天的主角。「给它的台词要多点」大家都知道它最初是LinkedIn研发的，在2011年初开源之后，2012年成为Apache 项目，直到今天 依然是Apache 顶级项目。一开始用ACtiveMQ 来解决业务问题，但扩展性不足，还经常导致阻塞，所以开始自研了Kafka。
• 2012年阿里的RocketMQ开源。它也参照Kafka的设计理念，并主要对消息的可靠传输及事务性做了优化。
• 2016年Yahoo开发的Pulsar，这是基于云原生的消息系统，在多租户上可以做彻底的隔离，处理速度也更快，但是相对来说落地实践的场景还不够丰富，并且缺少一个里程碑的稳定版本。
• 2018年，我们的JMQ完成第四次大版本的迭代，性能也有了极大的提升。也就是我们现在在用的JMQ4

大家看kafka后边几个的设计架构，也会多多少少能从其中看出一些kafka设计的影子。

一简介

kafka 设计规划

总所周知，kafka最初是美国一家LinkedIn 的工程师研发，当时主要解决数据管道（data pipeline）的问题。
当时这家公司内部有很多子系统用于收集和分析，比如用户行为操作，他们会定期把数据以xml的格式发送到统一的地方进行离线处理。既然现有解决方案
reps便开始组织团队进行消息传递系统的研发； 定位：作为统一的平台来处理大公司可能拥有的所有实时数据源。定位很宏观，所以要想的场景就比较多。
" 它必须具有高吞吐量才能支持高容量事件流，例如实时日志聚合。

“

• 它必须具有高吞吐量才能支持高容量事件流，例如实时日志聚合。据加载。
• 它需要优雅地处理大量数据积压，以便能够支持来自离线系统的定期数据加载。
• 这也意味着系统必须处理低延迟交付，以处理更传统的消息传递用例。在存在机器故障的情况下保证容错能力。"
• 在将流馈送到其他数据系统进行服务的情况下，我们知道系统必须能够在存在机器故障的情况下保证容错能力。
  ”

也就是用我们常说的：高吞吐，低延时，支持离线，高可用
那么kafka是怎么做的？这里面
有很多设计师自己的思考，篇幅原因，我们重点说几个核心内容。

开始之前就问一个问题，实现这些很难么？🌚
别急，那我们先看一下kafka出现之前的当时的消息系统架构。

rabbitMQ 架构分析

rabbitMQ 高可用模式：

这是rabbitMq的高可用模式。每个节点都有一个队列，生产者生产的数据投递到指定的服务器的队列上，然后队列再进行同步。每个节点都有一个完整的镜像，包含全部的数据。任何一个节点宕机之后，其他节点都还有一个完整数据，别的consumer 可以到其他节点上去消费数据。
但是当我们数据量再大的时候，是没办法水平扩展的。

kafka 架构图

水平扩展：
我们看kafka的这个图里，生产者ProducerA 生产的2条消息分别发送到了分区0 和分区1，然后消费者从对应的分区上进行消费。
如果topic的数据量更大的话，那还是可以增加新的分区来水平扩展。也就是可以增加新的分片数量。

高可用：
每个分区可以设置副本。Follower实时从 leader 中同步数据，保持和 leader 数据的同步。leader 发生故障时，某个 follower 会成为新的 leader 。消费者会通过zk感知到变化，然后去从新的leader上消费数据。

那是谁来控制副本选举的过程的呢？
整个集群会选举一个Broker 作为Controller。他负责管理整个集群所有分区和副本的状态调整。

在这里先有一个大体的印象，我们接下来具体看下主要模块的工作流程是什么样的。

二基本原理

1持久化

磁盘

有第一个问题要讨论的就是磁盘。因为传统观念是磁盘总是很慢。kafka重视高吞吐，为什么要选用慢的磁盘存储？

顺序写

实际上快慢完全取决于使用方式。这是09年发表的实验结论：

看前两条， 磁盘的顺序写入量是随机写入量的100多倍。

！！值得注意的是，这个是对内存的随机写，也就是说当我们顺序写入磁盘是，是可能比内存效率还要高。假设是如果基于内存的话，：对象内存开销非常高，并且也会随着堆内数据的增加，GC的速度越来越慢。

综合对比：
硬盘相对于内存来说，无论扩展还是成本稳定性优势更明显，可以保存更长，所以的存储方式选用了磁盘进行顺序写。

存储设计及查找过程

那么我们来看下，具体的存储设计。

【图1】生产者生产的消息落到这个分区之后，会顺序的追加到这里，每条数据都有自己的 offset。

【图2】每个分区对应于一个log 文件。为防止 log 文件过大导致数据定位效率低下，将每个 partition 分为多个 segment。每个 segment里对应这样几个文件。

【图3】这几个文件都以为当前 segment的第一条消息的 offset 命名

【图4】通过索引可以定位到对应的文件位置。

【小优化】这里有个优化是，为了减少文件大小，index采用稀疏索引，大约每往log文件写入4kb数据，会往index文件写入一条索引。那如果我想找图4 里，offset为2的，我就先通过索引1找到对应的日志文件，然后再去进行遍历。

从图里看到的问题来了：
a 消息顺序性（限单分区内 局部顺序）：
从这样的存储结构里，我们能看到消息在存储过程是有顺序的，但是只局限于这一个分区上。比如我发送了两条消息。一条消息到分区0 ，一条消息到分区1 那有顺序么，保证不了，因为不在一个队列里排序。

b IO竞争：
另一点上，因为顺序写，所以磁盘的效率非常高，但是如果单台机器上的Partition 数量太多，会发生多个文件同时IO的情况，会引起IO竞争，性能大幅下降。所以单机Partatio数量需要适度（就是具体是多少，这也不好说)。

对比

这里我们对比一下。这两个在存储上都有借鉴kafka，又都有不同的改造。

RocketMQ ：
在同一个Broker上，所有的数据（包括不同topic的数据）都记录在同一个文件中（CommitLog）。这样不就避免了IO的问题么？全部线性来写。这样效率高了，但是对比kafka来说，又灵活性低了。比如我想按照topic定制不同的存储时间。它就做不到了。

Jmq4 ：
比他们更晚，它在这一点取了两者的平衡。在同一个Broker上，按topic分类，然后同一个topic下对应一组消息文件（Log Files），顺序存放这个Topic的消息。然后收到的消息按照对应的Topic写入依次追加写入消息文件中，然后异步创建索引并写入对应Partition的索引文件中。

各有优劣，都是在灵活性和性能之间取一个平衡。

小结

2生产者

接下来说生产者这部分。看看生产者这边要考虑哪些事情：

第一问 发送到哪

首先消息发送到哪，这是生产者发送的时候就可以决定了。有这样几种方式。

• 指定分区
• Hash(key)%partition
• Round-robin

比如想保持相对的顺序性，可以按照一个维度去hash，保证比如同一个订单落到同一个分区里，这样大部分情况下就是有序的。

第二问 怎么知道发送成功了

大家说这个问题很简单。就发送之后有返回码呗。 集群给个回应不就行了。
是的。那这个回应是什么时候给呢？

从最简单的消息系统说起

等等，先心里想一个答案再接着看分析：

• A 发过去就完事了， 这样吞吐率最高。但是万一没成功 消息就丢了
• B leader 收到之后 ，follower还没来得及同步数据 。Leader 就挂掉了-》数据丢了
• C 半数以上。也就是这一半的机器没都挂掉，就还有数据。容灾能容这一半，这个看着很合理（实际上kafka没有这个选项）
• D 全部follower 同步之后 ，好处是 只要有一台机器没挂，消息就不会丢。但是缺点是如果有一台机器故障 一直没有收到呢？（实际上kafka也没有这个选项）

如果要保证可靠性，确保消息不丢失那答案是：

那什么是ISR列表：
ISR（In-sync-replicas）：在同步中的副本，维护机制是：如果在replica.lag.time.max.ms时间内系统没有发送fetch请求，或者已经在发送请求，但是在该限定时间内没有赶上Leader的数据就被剔除ISR列表。
（在Kafka-0.9.0版本剔除replica.lag.max.messages消息个数限定，因为这个会导致其他的Broker节点频繁的加入和退出ISR。）

想想有了这样的列表，是不是比固定的半数还是全部机器都要灵活和靠谱。实际上这个返回值什么时候返回，也是可配的，看场景：

好了，到这里，我们知道如果配置了 acks=all，那发送的消息就不会丢失了，
但是如果我这样设置了，可是给返回的时候机器挂了，那生产者还要重新发送一个消息，这样不就重复了么。那么消息可以避免重复么？

第三问 既不丢也不重复？

也就是语义上 正好一次。

嗯，kafka在这方面也是做了一些努力的！

这是什么意思呢？这个原理是什么呢？
我们首先来看下为什么会重复，

不重复的原理就是有一个唯一ID，然后这样当第二次落到分区的时候，发现一样了，就不再重新写入了

但是这样也有一个问题：问题再这个唯一ID，是由PID 和 SequenceNumber组成的。而PID是在服务启动的时候。也就是说重启PID就变了。那重启之后赶上消息重新发送，这时候消息就重复了。还有没有办法解决呢？
有。
kafka后续迭代的版本里增加了事务。当然事务不止是为了解决这个问题）

图不用细看 ==

先给结论：
幂等+事务，实现生产者恰好一次语义
怎么做到的呢？

为了实现事务，应用程序必须提供全局唯一的transactionalId,这通过客户端参数显示设置。
有了这个ID，那如果发送返回结果失败了，事务要么等恢复了继续原来的事务，我们这个事务回滚。恢复的话，因为事务ID可以直接对应到原来的消息ID，这样的话，我们就可以知道这个消息是重复的。 回滚的话，原来写入消息就回滚了，重新发送就行，也不会重复。

小结

3消费者

消费者这边，要考虑哪些事情呢？

第一问 Push or pull

第一个问题是 消息模型里两种经典方式，Push or pull 。
kafka 选用哪种方式。分析一下：
Push ：很难适应消费速率不同的消费者，因为消息发送速率是由 broker 决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成 consumer 来不及处理消息，拒绝服务。

pull ：可以根据 consumer 的消费能力以适当的速率消费消息。pull 模式不足之处是，如果 kafka 没有数据，消费者可能会陷入循环中，一直返回空数据。

【小优化】👉🏻 针对这一点，Kafka 的消费者在消费数据时会传入一个时长参数 timeout，如果当前没有数据可供消费，consumer 会等待一段时间之后再返回，这段时长即为 timeout。

第二问 消费哪个分区

我知道消息是拉取了，那我有个消费者，这么多分区，我该消费哪个呢？有什么规则么？有，这几种分区策略：

但是这里有一点需要注意并且很重要的是，因为kafka一个分区只能对应一组消费者里的一个。

【重要】所以单纯增加消费者数量到一定程度后无法增加消费速度。

比如这种情况 不论是哪种分配策略，都一定是只有2个消费者在同时工作。

第三问 从哪条消息开始消费

我们知道每个消息有个offset ，通过offset就能知道从哪消费。那么这个offset 应该是谁来维护呢，消费者还是生产者 还是哪里？这肯定不能是消费端，因为一旦机器重启，这消息就丢了。但是也需要消费者消费完之后说一声。这样服务端才知道你消费完了。

offset 提交机制有两种：

这就是我们常遇到这个问题。就是消费者到底是要自动提交offset 还是手动提交？分析一下：
a : 设置自动提交之后，间隔一段时间就提交一次告诉服务端我消费到这里了。但是问题是，如果你消费的慢，可能这时候这消息还没消费完。那机器故障消息就丢了。另外，我已经消费完了，但是还没到自动提交时间，机器宕机了。那服务端不知道你消费到这里了。那下次等重启之后，又重新消费了,重了。

1 自动提交 offset ——》 可能丢失，重复
2手动提交 offset ——》 不丢失，可能重复

小结

所以说，消费端这边是可以做到不丢失，但是不保证不重复，还是需要业务幂等。

三特性及其具体保证

第三部分是了解了基本原理之后，我们再来整体看下 kafka特性 以及对其都做了哪些设计

1高可用及水平扩展

这部分内容做个回顾，在上文架构图里有提。（略）

2高吞吐

我们具体看下高吞吐的保证：
1-4 也可以理解为保证低延迟 , 就是生产到消费的时间尽可能的短：
5-7 是尽可能的多。这样就比较好理解了。

1 直连

这有什么特点？因为kafka是直接通过生产端来进行了负载均衡，没有引入其他的负载均衡中间件。因为，确实是有其他MQ组件来单独用的负载均衡中间件。身份证号就不报了。感兴趣的可以去查查

2 顺序写

这点存储里已经提过啦。

3 PageCache

在 Kafka 中，大量使用了 PageCache， 当一个进程准备读取磁盘上的文件内容时，操作系统会先查看待读取的数据页是否在 PageCache 中，如果命中则直接返回数据，从而避免了对磁盘的 I/O 操作；如果没有命中，操作系统则会向磁盘发起读取请求并将读取的数据页存入 PageCache 中，之后再将数据返回给进程。

4 零拷贝

于Linux 系统调用机制，就是跳过“用户缓冲区”的拷贝，建立一个磁盘空间和内存的直接映射，数据不再复制到“用户态缓冲区”系统上下文切换减少为2次，可以提升至少一倍的性能

在实际应用中，如果我们需要把磁盘中的某个文件内容发送到远程服务器上，那么它必须要经过几个拷贝的过程：见图一
从磁盘中读取目标文件内容拷贝到内核缓冲区，CPU控制器再把内核缓冲区的数据复制到用户空间的缓冲区中；接着在应用程序中，把用户空间缓冲区中的数据拷贝到内核下的Socket Buffer中。
最后，把在内核模式下的SocketBuffer中的数据赋值到网卡缓冲区（NIC Buffer)；网卡缓冲区再把数据传输到目标服务器上。

在这个过程中我们可以发现，数据从磁盘到最终发送出去，要经历4次拷贝，而在这四次拷贝过程中，有两次拷贝是浪费的，而零拷贝，就是把这两次多于的拷贝省略掉，应用程序可以直接把磁盘中的数据从内核中直接传输给Socket，而不需要再经过应用程序所在的用户空间。
见上图二所示。

5 批量压缩

压缩也是为了减少文件的大小，注意是批量，一批进行统一的压缩。

1）kafka的发送端将消息按照批量（如果批量设置一条或者很小，可能有相反的效果）的方式进行压缩。
2）服务器端直接将压缩消息保存（如果kafka的版本不同，也存在broker需要先解压缩再压缩的问题）
3）消费端自动解压缩

kafka支持三种压缩算法，lz4、snappy、gzip。

6 批量发送

“避免过多的小 I/O 操作”
在很多情况下，系统的瓶颈不是CPU或磁盘，而是网络IO，所以降低大量小的 I/O：使消息消息分批发送会使整个吞吐量得到明细的提升。

7 水平扩展

一台机器的吞吐有限，我还可以用N台。

总结

今天主要写了一些kafka基本的原理及其高吞吐的保障。希望能对看到这里的同学有一些帮助 🤦🏻‍♀️ 。如果有哪里不对的地方，欢迎大家提出来~

然后，点个赞再走啊！！ 🌚

---

1 官方文档 kafka design: https://kafka.apache.org/documentation/#design

2 部分图引自网络。

---