卡夫卡是什么？

Kafka是基于Scala和Java语言开发的。它在设计中大量使用了批处理和异步的思想，最高可以每秒处理数百万条消息。它是一个用于构建实时数据管道和数据流的应用程序。

卡夫卡的应用场景

Kafka是一个分布式流媒体平台。该平台有三个关键功能:

消息队列：发布和订阅消息流，这个功能类似于消息队列，这也是Kafka被归类为消息队列的原因。容错的持久方式存储记录消息流：Kafka会把消息持久化到磁盘，有效避免消息丢失的风险。流式处理平台：在消息发布的时候进行处理，Kafka提供了一个完整的流式处理类库。

卡夫卡有两个主要的应用场景:

消息队列：建立实时流数据管道，可靠地在系统或应用程序之间获取数据。数据处理：构建实时的流数据处理程序来转换或处理数据流。

注:在2.8预览版中，Kafka采用了Raft元数据模式，取消了对Zookeeper的依赖。

卡夫卡的版本里程碑

版本号备注0.8引入了副本机制，成为了一个真正意义上完备的分布式高可靠消息队列解决方案0.8.2新版本 Producer API，即需要指定 Broker 地址的 Producer0.9增加了基础的安全认证 / 权限，Java 重写了新版本消费者 API0.10引入了 Kafka Streams0.11提供幂等性 Producer API 以及事务（Transaction） API，对 Kafka 消息格式做了重构。1.0Kafka Streams 的各种改进2.0Kafka Streams 的各种改进

卡夫卡的优点

高吞吐、低延时：这是 Kafka 显著的特点，Kafka 能够达到百万级的消息吞吐量，延迟可达毫秒级。持久化存储：Kafka 的消息最终持久化保存在磁盘之上，提供了顺序读写以保证性能，并且通过 Kafka 的副本机制提高了数据可靠性。分布式可扩展：Kafka的数据是分布式存储在不同broker节点的，以topic组织数据并且按Partition进行分布式存储，整体的扩展性都非常好。高容错性：集群中任意一个 broker 节点宕机，Kafka 仍能对外提供服务。

卡夫卡基本结构

Kafka有四个核心API:

Producer API： 发布消息到1个或多个topic（主题）中。Consumer API：来订阅一个或多个topic，并处理产生的消息。Streams API：充当一个流处理器，从1个或多个topic消费输入流，并生产一个输出流到1个或多个输出topic，有效地将输入流转换到输出流。Connector API：可构建或运行可重用的生产者或消费者，将topic连接到现有的应用程序或数据系统。例如，连接到关系数据库的连接器可以捕获表的每个变更。

卡夫卡的关键词

Producer：消息和数据的生产者，向Kafka的一个Topic发布消息的进程/代码/服务。Consumer：消息和数据的消费者，订阅数据（Topic）并且处理发布的消息的进程/代码/服务。Consumer Group：对于同一个Topic，会广播给不同的Group。在一个Group中，一条消息只能被消费组中一个的Consumer消费。

使用者组中的使用者不能多于分区，否则多余的使用者将在空等待，并且不会收到消息。

Topic：每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。作用是对数据进行区分、隔离。Broker：Kafka集群中的每个Kafka节点。保存Topic的一个或多个Partition。Partition：物理概念，Kafka下数据储存的基本单元。一个Topic数据，会被分散存储到多个Partition，每一个Partition都是一个顺序的、不可变的消息队列，并且可以持续的添加消息。

注：话题:每一条发布到Kafka集群的消息都有一个类别，叫做话题。功能是区分和隔离数据。broker:Kafka集群中的每个Kafka节点。保存一个或多个主题分区。分区:物理概念，卡夫卡下数据存储的基本单位。一个主题数据会被存储在多个分区中，每个分区都是一个顺序的、不可变的消息队列，并且消息可以不断地添加。注意:

每个主题的信息被分成多个分区。如果分区的数量设置为1，则可以保证消息消费的顺序。如果一个主题有n个分区，一个集群有n个代理，那么每个代理存储该主题的一个分区。如果一个主题有N个分区，一个集群有(N+M)个代理，那么N个代理存储主题的一个分区，剩下的M个代理不存储主题的分区数据。如果一个主题有n个分区，并且集群中的代理数量小于n，那么一个代理存储该主题的一个或多个分区。在实际生产环境中，尽量避免这种情况，容易导致Kafka集群数据不平衡。当代理收到消息时，它会根据分区算法选择将消息存储在哪个分区中。其路由机制是根据指定的分区优先路由；如果你没有指定一个部分，但是指定了一个键，散列这个键的值来选择一个部分；如果既没有指定分区也没有指定键，则通过轮询选择分区。

Offset：偏移量，分区中的消息位置，由Kafka自身维护，Consumer消费时也要保存一份Offset以维护消费过的消息位置。Replication：同一个Partition可能会有多个副本，多个副本之间数据是一样的，增加容错性与可扩展性。

注意:

当集群中的代理挂起时，系统可以主动使用***来提供服务。每个题目的***系数默认设置为1，可以在创建题目时单独设置。***的基本单位是主题的划分。所有的读写都是由领导进行的，跟随者只是作为数据备份。追随者必须能够及时***领导者的数据。

Replication Leader：一个Partition的多个副本上，需要一个Leader负责该Partition上与Producer和Consumer交互。一个Partition只对应一个Replication Leader。Replication Follower：Follower跟随Leader，所有写请求都会广播给所有Follower，Follower与Leader保持数据同步。ReplicaManager：负责管理当前Broker所有分区和副本的信息，处理KafkaController发起的一些请求，副本状态的切换、添加/读取消息等。Rebalance。消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程。Rebalance是Kafka消费者端实现高可用的重要手段。

Kafka通过Zookeeper管理集群配置，选举领导者，并在消费者群体发生变化时重新平衡。

卡夫卡的***机制

如何将所有***均匀分布到整个集群

为了更好地平衡负载，Kafka试图将所有分区平均分配给整个集群。典型的部署方法是一个主题的分区数量大于代理数量。同时，为了提高Kafka的容错能力，也需要将同一个分区的***尽可能的分布到不同的机器上。如果所有***都在同一个代理上，那么一旦代理关闭，分区的所有***都将无法工作，也就无法达到HA的效果。同时，如果一个代理关闭，需要确保它的负载可以平均分配给所有其他幸存的代理。

Kafka分配***的算法如下:

将所有Broker（假设共n个Broker）和待分配的Partition排序。将第i个Partition分配到第（i % n）个Broker上。将第i个Partition的第j个Replication分配到第（(i + j) % n）个Broker上。

HW高水位和LEO

HW是High Water***rk的缩写，俗称高水印。它标识了一个特定的消息偏移量，使用者只能提取该偏移量之前的消息。

如图所示，它代表一个日志文件。该日志文件中有9条消息。第一条消息的偏移量为0，最后一条消息的偏移量为8，偏移量为9的消息用虚线框表示，表示下一条要写入的消息。日志的HW是6，这意味着消费者只能拉取偏移量在0到5之间的消息，而偏移量为6的消息对消费者是不可见的。

是LEO Log End Offset的缩写，标识当前日志文件中要写入的下一条消息的偏移量。图中偏移量为9的位置是当前日志文件的LEO，LEO的大小相当于当前日志分区中最后一条消息的偏移量值加1。一个分区的ISR集中的每个副本都会维护自己的LEO，ISR集中最小的LEO就是该分区的HW，所以消费者只能消费HW之前的消息。

ISR副本集

ISR的全称是“In-Sync Replicas”，是与主副本同步的分区中的***列表。正常情况下，ISR必须包含Leader的副本。

ISR列表在Zookeeper中是持久的，ISR列表中的任何副本都有资格参与首领选举。

ISR的列表是动态的，副本被包含在ISR列表中的条件由参数replica.lag.time.***x.ms控制，该参数表示副本同步滞后于Leader的最大时间间隔，默认为10s，这意味着如果一个Follower所在的Broker由于JVM FullGC等问题相对于Leader延迟超过10s，就会被排除在ISR之外。卡夫卡就是这样设计的，主要是为了减少信息的丢失。只有与领袖副本实时同步的追随者副本才有资格参与领袖选举，也就是说相对实时。

注意:

分区中的所有副本统称为AR(分配副本)。ISR***是AR***的子集。与前导副本同步滞后太多的副本(不包括前导副本)构成OSR(不同步副本)。

***机制

如图所示，假设一个分区的ISR集中有三个副本，即一个主副本和两个从副本。此时分区的LEO和HW都是3。消息3和消息4将在从生产者发送后存储在领导者副本中。

在消息写入Leader副本之后，Follower副本会发送拉取请求来拉取消息3和消息4以进行消息同步。在消息被写入领导者副本之后，从者副本发送pull请求到pull消息3和消息4以进行消息同步。

在同步过程中，不同的跟随者副本具有不同的同步效率。在某个时刻，跟随者1完全跟上领导者副本，跟随者2仅同步消息3，因此领导者副本的LEO是5，跟随者1的LEO是5，跟随者2的LEO是4，则当前分区的HW取最小值4。此时，消费者可以使用偏移量在0到3之间的消息。

当所有副本都成功写入消息3和消息4时，整个分区的HW和LEO都变成了5，所以消费者可以消费偏移量为4的消息。

阅读与写作的分离

Kafka不支持读写分区。生产端和消费端的所有读写请求都由***领导者副本处理。***跟随者副本的主要工作是从领导者副本异步拉取消息，同步消息数据，不提供外部读写服务。

Kafka的设计是为了保证读写的一致性，因为拷贝同步是一个异步的过程。如果跟随者副本与领导者不完全同步，从跟随者副本读取数据可能无法读取最新消息。

卡夫卡的信息传递机制

Producer使用push模式向Broker发布消息，每条消息都被追加到专利中，属于顺序磁盘写入(顺序磁盘写入比随机内存写入效率更高，保证了kafka的吞吐量)。

生产者编写消息的顺序图如下:

流程描述:

Producer先从Zookeeper的”/brokers/…/state”节点找到该Partition的Leader。Producer将消息发送给该Leader。Leader将消息写入本地log。followers从Leader pull消息，写入本地log后Leader发送ACK。Leader收到所有ISR中的replica的ACK后，增加HW并向Producer发送ACK。

代理保存消息

每个分区实际上对应一个文件夹(该文件夹存储该分区的所有邮件和索引文件)。

无论是否被消费，卡夫卡都会保留所有的信息。删除旧数据有两种策略:

基于时间:log.retention.hours=168基于大小:log.retention.bytes=1073741824

消费者消费新闻

Kafka cluster保留所有消息，直到它们过期(无论消息是否被消费)。实际上，消费者持有的唯一元数据是这个偏移量，这意味着偏移量由消费者控制:通常，当消费者消费消息时，偏移量线性增加。但是实际的偏移量是由消费者控制的，消费者可以将偏移量重置到更早的位置，并重新读取消息。可以看出，这样的设计便于消费者操作，一个消费者的操作不会影响其他消费者对这个日志的处理。