Kafka 高吞吐量性能揭秘

共计 10421 个字符，预计需要花费 27 分钟才能阅读完成。

kafka 作为时下最流行的开源消息系统，被广泛地应用在数据缓冲、异步通信、汇集日志、系统解耦等方面。相比较于 RocketMQ 等其他常见消息系统，Kafka 在保障了大部分功能特性的同时，还提供了超一流的读写性能。

本文将针对 Kafka 性能方面进行简单分析，首先简单介绍一下 Kafka 的架构和涉及到的名词：

1. Topic：用于划分 Message 的逻辑概念，一个 Topic 可以分布在多个 Broker 上。
2. Partition：是 Kafka 中横向扩展和一切并行化的基础，每个 Topic 都至少被切分为 1 个 Partition。
3. Offset：消息在 Partition 中的编号，编号顺序不跨 Partition。
4. Consumer：用于从 Broker 中取出 / 消费 Message。
5. Producer：用于往 Broker 中发送 / 生产 Message。
6. Replication：Kafka 支持以 Partition 为单位对 Message 进行冗余备份，每个 Partition 都可以配置至少 1 个 Replication(当仅 1 个 Replication 时即仅该 Partition 本身)。
7. Leader：每个 Replication 集合中的 Partition 都会选出一个唯一的 Leader，所有的读写请求都由 Leader 处理。其他 Replicas 从 Leader 处把数据更新同步到本地，过程类似大家熟悉的 MySQL 中的 Binlog 同步。
8. Broker：Kafka 中使用 Broker 来接受 Producer 和 Consumer 的请求，并把 Message 持久化到本地磁盘。每个 Cluster 当中会选举出一个 Broker 来担任 Controller，负责处理 Partition 的 Leader 选举，协调 Partition 迁移等工作。
9. ISR(In-Sync Replica)：是 Replicas 的一个子集，表示目前 Alive 且与 Leader 能够“Catch-up”的 Replicas 集合。由于读写都是首先落到 Leader 上，所以一般来说通过同步机制从 Leader 上拉取数据的 Replica 都会和 Leader 有一些延迟(包括了延迟时间和延迟条数两个维度)，任意一个超过阈值都会把该 Replica 踢出 ISR。每个 Partition 都有它自己独立的 ISR。

以上几乎是我们在使用 Kafka 的过程中可能遇到的所有名词，同时也无一不是最核心的概念或组件，感觉到从设计本身来说，Kafka 还是足够简洁的。这次本文围绕 Kafka 优异的吞吐性能，逐个介绍一下其设计与实现当中所使用的各项“黑科技”。

Broker
不同于 Redis 和 MemcacheQ 等内存消息队列，Kafka 的设计是把所有的 Message 都要写入速度低容量大的硬盘，以此来换取更强的存储能力。实际上，Kafka 使用硬盘并没有带来过多的性能损失，“规规矩矩”的抄了一条“近道”。

首先，说“规规矩矩”是因为 Kafka 在磁盘上只做 Sequence I/O，由于消息系统读写的特殊性，这并不存在什么问题。关于磁盘 I / O 的性能，引用一组 Kafka 官方给出的测试数据(Raid-5，7200rpm)：

Sequence I/O: 600MB/s
Random I/O: 100KB/s

所以通过只做 Sequence I/ O 的限制，规避了磁盘访问速度低下对性能可能造成的影响。

接下来我们再聊一聊 kafka 是如何“抄近道的”。

首先，Kafka 重度依赖底层操作系统提供的 PageCache 功能。当上层有写操作时，操作系统只是将数据写入 PageCache，同时标记 Page 属性为 Dirty。当读操作发生时，先从 PageCache 中查找，如果发生缺页才进行磁盘调度，最终返回需要的数据。实际上 PageCache 是把尽可能多的空闲内存都当做了磁盘缓存来使用。同时如果有其他进程申请内存，回收 PageCache 的代价又很小，所以现代的 OS 都支持 PageCache。
使用 PageCache 功能同时可以避免在 JVM 内部缓存数据，JVM 为我们提供了强大的 GC 能力，同时也引入了一些问题不适用与 Kafka 的设计。

相关阅读：

分布式发布订阅消息系统 Kafka 架构设计 http://www.linuxidc.com/Linux/2013-11/92751.htm

Apache Kafka 代码实例 http://www.linuxidc.com/Linux/2013-11/92754.htm

Apache Kafka 教程笔记 http://www.linuxidc.com/Linux/2014-01/94682.htm

Apache kafka 原理与特性(0.8V)  http://www.linuxidc.com/Linux/2014-09/107388.htm

Kafka 部署与代码实例  http://www.linuxidc.com/Linux/2014-09/107387.htm

Kafka 介绍和集群环境搭建  http://www.linuxidc.com/Linux/2014-09/107382.htm

 
• 如果在 Heap 内管理缓存，JVM 的 GC 线程会频繁扫描 Heap 空间，带来不必要的开销。如果 Heap 过大，执行一次 Full GC 对系统的可用性来说将是极大的挑战。
• 所有在在 JVM 内的对象都不免带有一个 Object Overhead(千万不可小视)，内存的有效空间利用率会因此降低。
• 所有的 In-Process Cache 在 OS 中都有一份同样的 PageCache。所以通过只在 PageCache 中做缓存至少可以提高一倍的缓存空间。
• 如果 Kafka 重启，所有的 In-Process Cache 都会失效，而 OS 管理的 PageCache 依然可以继续使用。

PageCache 还只是第一步，Kafka 为了进一步的优化性能还采用了 Sendfile 技术。在解释 Sendfile 之前，首先介绍一下传统的网络 I / O 操作流程，大体上分为以下 4 步。

1. OS 从硬盘把数据读到内核区的 PageCache。
2. 用户进程把数据从内核区 Copy 到用户区。
3. 然后用户进程再把数据写入到 Socket，数据流入内核区的 Socket Buffer 上。
4. OS 再把数据从 Buffer 中 Copy 到网卡的 Buffer 上，这样完成一次发送。

整个过程共经历两次 Context Switch，四次 System Call。同一份数据在内核 Buffer 与用户 Buffer 之间重复拷贝，效率低下。其中 2、3 两步没有必要，完全可以直接在内核区完成数据拷贝。这也正是 Sendfile 所解决的问题，经过 Sendfile 优化后，整个 I / O 过程就变成了下面这个样子。

通过以上的介绍不难看出，Kafka 的设计初衷是尽一切努力在内存中完成数据交换，无论是对外作为一整个消息系统，或是内部同底层操作系统的交互。如果 Producer 和 Consumer 之间生产和消费进度上配合得当，完全可以实现数据交换零 I /O。这也就是我为什么说 Kafka 使用“硬盘”并没有带来过多性能损失的原因。下面是我在生产环境中采到的一些指标。
(20 Brokers, 75 Partitions per Broker, 110k msg/s)

更多详情见请继续阅读下一页的精彩内容：http://www.linuxidc.com/Linux/2016-03/129065p2.htm

通过以上的介绍不难看出，Kafka 的设计初衷是尽一切努力在内存中完成数据交换，无论是对外作为一整个消息系统，或是内部同底层操作系统的交互。如果 Producer 和 Consumer 之间生产和消费进度上配合得当，完全可以实现数据交换零 I /O。这也就是我为什么说 Kafka 使用“硬盘”并没有带来过多性能损失的原因。下面是我在生产环境中采到的一些指标。
(20 Brokers, 75 Partitions per Broker, 110k msg/s)

此时的集群只有写，没有读操作。10M/ s 左右的 Send 的流量是 Partition 之间进行 Replicate 而产生的。从 recv 和 writ 的速率比较可以看出，写盘是使用 Asynchronous+Batch 的方式，底层 OS 可能还会进行磁盘写顺序优化。而在有 Read Request 进来的时候分为两种情况，第一种是内存中完成数据交换。

Send 流量从平均 10M/ s 增加到了到平均 60M/s，而磁盘 Read 只有不超过 50KB/s。PageCache 降低磁盘 I / O 效果非常明显。

接下来是读一些收到了一段时间，已经从内存中被换出刷写到磁盘上的老数据。

其他指标还是老样子，而磁盘 Read 已经飚高到 40+MB/s。此时全部的数据都已经是走硬盘了(对硬盘的顺序读取 OS 层会进行 Prefill PageCache 的优化)。依然没有任何性能问题。

Tips
1. Kafka 官方并不建议通过 Broker 端的 log.flush.interval.messages 和 log.flush.interval.ms 来强制写盘，认为数据的可靠性应该通过 Replica 来保证，而强制 Flush 数据到磁盘会对整体性能产生影响。
2. 可以通过调整 /proc/sys/vm/dirty_background_ratio 和 /proc/sys/vm/dirty_ratio 来调优性能。
a. 脏页率超过第一个指标会启动 pdflush 开始 Flush Dirty PageCache。
b. 脏页率超过第二个指标会阻塞所有的写操作来进行 Flush。
c. 根据不同的业务需求可以适当的降低 dirty_background_ratio 和提高 dirty_ratio。

Partition
Partition 是 Kafka 可以很好的横向扩展和提供高并发处理以及实现 Replication 的基础。

扩展性方面。首先，Kafka 允许 Partition 在集群内的 Broker 之间任意移动，以此来均衡可能存在的数据倾斜问题。其次，Partition 支持自定义的分区算法，例如可以将同一个 Key 的所有消息都路由到同一个 Partition 上去。同时 Leader 也可以在 In-Sync 的 Replica 中迁移。由于针对某一个 Partition 的所有读写请求都是只由 Leader 来处理，所以 Kafka 会尽量把 Leader 均匀的分散到集群的各个节点上，以免造成网络流量过于集中。

并发方面。任意 Partition 在某一个时刻只能被一个 Consumer Group 内的一个 Consumer 消费(反过来一个 Consumer 则可以同时消费多个 Partition)，Kafka 非常简洁的 Offset 机制最小化了 Broker 和 Consumer 之间的交互，这使 Kafka 并不会像同类其他消息队列一样，随着下游 Consumer 数目的增加而成比例的降低性能。此外，如果多个 Consumer 恰巧都是消费时间序上很相近的数据，可以达到很高的 PageCache 命中率，因而 Kafka 可以非常高效的支持高并发读操作，实践中基本可以达到单机网卡上限。

不过，Partition 的数量并不是越多越好，Partition 的数量越多，平均到每一个 Broker 上的数量也就越多。考虑到 Broker 宕机 (Network Failure, Full GC) 的情况下，需要由 Controller 来为所有宕机的 Broker 上的所有 Partition 重新选举 Leader，假设每个 Partition 的选举消耗 10ms，如果 Broker 上有 500 个 Partition，那么在进行选举的 5s 的时间里，对上述 Partition 的读写操作都会触发 LeaderNotAvailableException。

再进一步，如果挂掉的 Broker 是整个集群的 Controller，那么首先要进行的是重新任命一个 Broker 作为 Controller。新任命的 Controller 要从 Zookeeper 上获取所有 Partition 的 Meta 信息，获取每个信息大概 3 -5ms，那么如果有 10000 个 Partition 这个时间就会达到 30s-50s。而且不要忘记这只是重新启动一个 Controller 花费的时间，在这基础上还要再加上前面说的选举 Leader 的时间 -_-!!!!!!

此外，在 Broker 端，对 Producer 和 Consumer 都使用了 Buffer 机制。其中 Buffer 的大小是统一配置的，数量则与 Partition 个数相同。如果 Partition 个数过多，会导致 Producer 和 Consumer 的 Buffer 内存占用过大。

tips
1. Partition 的数量尽量提前预分配，虽然可以在后期动态增加 Partition，但是会冒着可能破坏 Message Key 和 Partition 之间对应关系的风险。
2. Replica 的数量不要过多，如果条件允许尽量把 Replica 集合内的 Partition 分别调整到不同的 Rack。
3. 尽一切努力保证每次停 Broker 时都可以 Clean Shutdown，否则问题就不仅仅是恢复服务所需时间长，还可能出现数据损坏或其他很诡异的问题。

Producer
Kafka 的研发团队表示在 0.8 版本里用 Java 重写了整个 Producer，据说性能有了很大提升。我还没有亲自对比试用过，这里就不做数据对比了。本文结尾的扩展阅读里提到了一套我认为比较好的对照组，有兴趣的同学可以尝试一下。

其实在 Producer 端的优化大部分消息系统采取的方式都比较单一，无非也就化零为整、同步变异步这么几种。

Kafka 系统默认支持 MessageSet，把多条 Message 自动地打成一个 Group 后发送出去，均摊后拉低了每次通信的 RTT。而且在组织 MessageSet 的同时，还可以把数据重新排序，从爆发流式的随机写入优化成较为平稳的线性写入。

此外，还要着重介绍的一点是，Producer 支持 End-to-End 的压缩。数据在本地压缩后放到网络上传输，在 Broker 一般不解压(除非指定要 Deep-Iteration)，直至消息被 Consume 之后在客户端解压。

当然用户也可以选择自己在应用层上做压缩和解压的工作(毕竟 Kafka 目前支持的压缩算法有限，只有 GZIP 和 Snappy)，不过这样做反而会意外的降低效率！！！！Kafka 的 End-to-End 压缩与 MessageSet 配合在一起工作效果最佳，上面的做法直接割裂了两者间联系。至于道理其实很简单，压缩算法中一条基本的原理“重复的数据量越多，压缩比越高”。无关于消息体的内容，无关于消息体的数量，大多数情况下输入数据量大一些会取得更好的压缩比。

不过 Kafka 采用 MessageSet 也导致在可用性上一定程度的妥协。每次发送数据时，Producer 都是 send()之后就认为已经发送出去了，但其实大多数情况下消息还在内存的 MessageSet 当中，尚未发送到网络，这时候如果 Producer 挂掉，那就会出现丢数据的情况。

为了解决这个问题，Kafka 在 0.8 版本的设计借鉴了网络当中的 ack 机制。如果对性能要求较高，又能在一定程度上允许 Message 的丢失，那就可以设置 request.required.acks=0 来关闭 ack，以全速发送。如果需要对发送的消息进行确认，就需要设置 request.required.acks 为 1 或 -1，那么 1 和 - 1 又有什么区别呢？这里又要提到前面聊的有关 Replica 数量问题。如果配置为 1，表示消息只需要被 Leader 接收并确认即可，其他的 Replica 可以进行异步拉取无需立即进行确认，在保证可靠性的同时又不会把效率拉得很低。如果设置为 -1，表示消息要 Commit 到该 Partition 的 ISR 集合中的所有 Replica 后，才可以返回 ack，消息的发送会更安全，而整个过程的延迟会随着 Replica 的数量正比增长，这里就需要根据不同的需求做相应的优化。

tips
1. Producer 的线程不要配置过多，尤其是在 Mirror 或者 Migration 中使用的时候，会加剧目标集群 Partition 消息乱序的情况(如果你的应用场景对消息顺序很敏感的话)。
2. 0.8 版本的 request.required.acks 默认是 0(同 0.7)。

Consumer
Consumer 端的设计大体上还算是比较常规的。

• 通过 Consumer Group，可以支持生产者消费者和队列访问两种模式。
• Consumer API 分为 High level 和 Low level 两种。前一种重度依赖 Zookeeper，所以性能差一些且不自由，但是超省心。第二种不依赖 Zookeeper 服务，无论从自由度和性能上都有更好的表现，但是所有的异常 (Leader 迁移、Offset 越界、Broker 宕机等) 和 Offset 的维护都需要自行处理。
• 大家可以关注下不日发布的 0.9 Release。这帮货又用 Java 重写了一套 Consumer。把两套 API 合并在一起，同时去掉了对 Zookeeper 的依赖。据说性能有大幅度提升哦~~

tips
强烈推荐使用 Low level API，虽然繁琐一些，但是目前只有这个 API 可以对 Error 数据进行自定义处理，尤其是处理 Broker 异常或由于 Unclean Shutdown 导致的 Corrupted Data 时，否则无法 Skip 只能等着“坏消息”在 Broker 上被 Rotate 掉，在此期间该 Replica 将会一直处于不可用状态。

扩展阅读
Sendfile: https://www.ibm.com/developerworks/cn/java/j-zerocopy/
So what’s wrong with 1975 programming: https://www.varnish-cache.org/trac/wiki/ArchitectNotes
Benchmarking: https://engineering.linkedin.com/kafka/benchmarking-apache-kafka-2-million-writes-second-three-cheap-machines

Kafka 的详细介绍：请点这里
Kafka 的下载地址：请点这里

本文永久更新链接地址：http://www.linuxidc.com/Linux/2016-03/129065.htm

kafka 作为时下最流行的开源消息系统，被广泛地应用在数据缓冲、异步通信、汇集日志、系统解耦等方面。相比较于 RocketMQ 等其他常见消息系统，Kafka 在保障了大部分功能特性的同时，还提供了超一流的读写性能。

本文将针对 Kafka 性能方面进行简单分析，首先简单介绍一下 Kafka 的架构和涉及到的名词：

1. Topic：用于划分 Message 的逻辑概念，一个 Topic 可以分布在多个 Broker 上。
2. Partition：是 Kafka 中横向扩展和一切并行化的基础，每个 Topic 都至少被切分为 1 个 Partition。
3. Offset：消息在 Partition 中的编号，编号顺序不跨 Partition。
4. Consumer：用于从 Broker 中取出 / 消费 Message。
5. Producer：用于往 Broker 中发送 / 生产 Message。
6. Replication：Kafka 支持以 Partition 为单位对 Message 进行冗余备份，每个 Partition 都可以配置至少 1 个 Replication(当仅 1 个 Replication 时即仅该 Partition 本身)。
7. Leader：每个 Replication 集合中的 Partition 都会选出一个唯一的 Leader，所有的读写请求都由 Leader 处理。其他 Replicas 从 Leader 处把数据更新同步到本地，过程类似大家熟悉的 MySQL 中的 Binlog 同步。
8. Broker：Kafka 中使用 Broker 来接受 Producer 和 Consumer 的请求，并把 Message 持久化到本地磁盘。每个 Cluster 当中会选举出一个 Broker 来担任 Controller，负责处理 Partition 的 Leader 选举，协调 Partition 迁移等工作。
9. ISR(In-Sync Replica)：是 Replicas 的一个子集，表示目前 Alive 且与 Leader 能够“Catch-up”的 Replicas 集合。由于读写都是首先落到 Leader 上，所以一般来说通过同步机制从 Leader 上拉取数据的 Replica 都会和 Leader 有一些延迟(包括了延迟时间和延迟条数两个维度)，任意一个超过阈值都会把该 Replica 踢出 ISR。每个 Partition 都有它自己独立的 ISR。

以上几乎是我们在使用 Kafka 的过程中可能遇到的所有名词，同时也无一不是最核心的概念或组件，感觉到从设计本身来说，Kafka 还是足够简洁的。这次本文围绕 Kafka 优异的吞吐性能，逐个介绍一下其设计与实现当中所使用的各项“黑科技”。

Broker
不同于 Redis 和 MemcacheQ 等内存消息队列，Kafka 的设计是把所有的 Message 都要写入速度低容量大的硬盘，以此来换取更强的存储能力。实际上，Kafka 使用硬盘并没有带来过多的性能损失，“规规矩矩”的抄了一条“近道”。

首先，说“规规矩矩”是因为 Kafka 在磁盘上只做 Sequence I/O，由于消息系统读写的特殊性，这并不存在什么问题。关于磁盘 I / O 的性能，引用一组 Kafka 官方给出的测试数据(Raid-5，7200rpm)：

Sequence I/O: 600MB/s
Random I/O: 100KB/s

所以通过只做 Sequence I/ O 的限制，规避了磁盘访问速度低下对性能可能造成的影响。

接下来我们再聊一聊 kafka 是如何“抄近道的”。

首先，Kafka 重度依赖底层操作系统提供的 PageCache 功能。当上层有写操作时，操作系统只是将数据写入 PageCache，同时标记 Page 属性为 Dirty。当读操作发生时，先从 PageCache 中查找，如果发生缺页才进行磁盘调度，最终返回需要的数据。实际上 PageCache 是把尽可能多的空闲内存都当做了磁盘缓存来使用。同时如果有其他进程申请内存，回收 PageCache 的代价又很小，所以现代的 OS 都支持 PageCache。
使用 PageCache 功能同时可以避免在 JVM 内部缓存数据，JVM 为我们提供了强大的 GC 能力，同时也引入了一些问题不适用与 Kafka 的设计。

相关阅读：

分布式发布订阅消息系统 Kafka 架构设计 http://www.linuxidc.com/Linux/2013-11/92751.htm

Apache Kafka 代码实例 http://www.linuxidc.com/Linux/2013-11/92754.htm

Apache Kafka 教程笔记 http://www.linuxidc.com/Linux/2014-01/94682.htm

Apache kafka 原理与特性(0.8V)  http://www.linuxidc.com/Linux/2014-09/107388.htm

Kafka 部署与代码实例  http://www.linuxidc.com/Linux/2014-09/107387.htm

Kafka 介绍和集群环境搭建  http://www.linuxidc.com/Linux/2014-09/107382.htm

 
• 如果在 Heap 内管理缓存，JVM 的 GC 线程会频繁扫描 Heap 空间，带来不必要的开销。如果 Heap 过大，执行一次 Full GC 对系统的可用性来说将是极大的挑战。
• 所有在在 JVM 内的对象都不免带有一个 Object Overhead(千万不可小视)，内存的有效空间利用率会因此降低。
• 所有的 In-Process Cache 在 OS 中都有一份同样的 PageCache。所以通过只在 PageCache 中做缓存至少可以提高一倍的缓存空间。
• 如果 Kafka 重启，所有的 In-Process Cache 都会失效，而 OS 管理的 PageCache 依然可以继续使用。

PageCache 还只是第一步，Kafka 为了进一步的优化性能还采用了 Sendfile 技术。在解释 Sendfile 之前，首先介绍一下传统的网络 I / O 操作流程，大体上分为以下 4 步。

1. OS 从硬盘把数据读到内核区的 PageCache。
2. 用户进程把数据从内核区 Copy 到用户区。
3. 然后用户进程再把数据写入到 Socket，数据流入内核区的 Socket Buffer 上。
4. OS 再把数据从 Buffer 中 Copy 到网卡的 Buffer 上，这样完成一次发送。

整个过程共经历两次 Context Switch，四次 System Call。同一份数据在内核 Buffer 与用户 Buffer 之间重复拷贝，效率低下。其中 2、3 两步没有必要，完全可以直接在内核区完成数据拷贝。这也正是 Sendfile 所解决的问题，经过 Sendfile 优化后，整个 I / O 过程就变成了下面这个样子。

通过以上的介绍不难看出，Kafka 的设计初衷是尽一切努力在内存中完成数据交换，无论是对外作为一整个消息系统，或是内部同底层操作系统的交互。如果 Producer 和 Consumer 之间生产和消费进度上配合得当，完全可以实现数据交换零 I /O。这也就是我为什么说 Kafka 使用“硬盘”并没有带来过多性能损失的原因。下面是我在生产环境中采到的一些指标。
(20 Brokers, 75 Partitions per Broker, 110k msg/s)

更多详情见请继续阅读下一页的精彩内容：http://www.linuxidc.com/Linux/2016-03/129065p2.htm