一、Kafka

Kafka是一个分布式的消息系统。

二、解决问题

消息系统通常被应用于异步处理、应用解耦、流量削峰、消息通信等场景。 异步处理 生产者将消息写入消息队列中，消费者异步拉取消息队列消息，从而提升消息处理能力。 应用解耦 Kafka作为消息传递的媒介，各子系统只需要做系统责任内的事情。生产者-消费者模式，Kafka就是消息队列。 流量削峰 正常情况下，上游服务（如报价、营销等）常年流量较大，面对大流量时能够较为从容地应对，但下游应用（如：交易、订单等）由于常年流量较小，面对大流量时会因为准备不足，而导致系统被打垮，引发雪崩。 为了应对这一问题，可以利用消息队列作为临时数据存储节点，消费者根据自身消费能力，通过拉取的方式控制消费速度，达到流量削峰的目的。

三、特性

读写效率 Kafka在面对大流量数据时，能够高效地处理消息的存储与查询。通过软件设计避免硬件读取磁盘的性能瓶颈。

网络传输 批量读取消息，对消息进行批量压缩，从而提升网络利用率。

并发能力 Kafka支持消息分区，每个分区内保证消息的顺序性，多分区之间能够支持并发操作，提升Kafka并发操作。

持久化能力 Kafka将消息持久化至硬盘。网络传输不可靠，所以需要将数据进行持久化。其中利用了零拷贝、顺序读、顺序写、页缓存等技术使Kafa具备高吞吐特性。

可靠性 支持分区多副本，Leader副本负责读写，Follow副本只负责同步Leader副本数据，实现消息冗余备份，提升Kafka容灾能力。

水平扩展 多Producer、**Broker、Consumer，均为分布式，多Consumer可以加入同一Consumer Group，每个分区只能分配一个Consumer，当Kafka服务端增加分区数量进行水平扩展时，可以向Consumer Group添加Consumer，提升消费能力。当Consumer Group中有Consumer出现故障下线时，能通过再平衡（Rebalance）对分区进行再分配。

四、基本概念

消息&批次

消息 （1）消息是Kafka的基本单位； （2）消息由key和value的byte数组构成; （3）key能够根据策略将消息发送到指定分区。 批次 （1）为了提升效率，消息被分批写入kafka，同一组消息必须属于同一主题的同一分区； （2）分批发送能够降低网络开销，提升传输速度。

主题&分区

主题（Topic）是用于存储消息分类关系的逻辑单元，可以看做存储消息的集合。分区（partition）是Kafka数据存储的基本单元，可以看做存储消息的集合的子集。Kafka消息通过主题进行分类，同一Topic的不同分区（partition）会分配在不用的Broker上，分区机制提供横向扩展的基础，可以通过增加并在其上分配partition来提升Kafka的消息并行处理能力。

日志

Log基本概念 （1）分区逻辑上对应一个Log，生产者将消息写入分区实际是写入分区对应的Log； （2）Log可以对应磁盘上的文件夹，其由多个Segment组成，每个Segment对应一个日志文件和索引文件； （3）当Segment大小超出限制时，就会创建新的Segment； （4）Kafka采用顺序I/O，所以只会向最新的Segment追加数据； （5）索引采用稀疏索引，运行时将其映射至内存中，提升索引速度。

Log保存与压缩

日志保存 （1）时间限制 根据保留时间，当消息在kafka中保存的时间超过指定时间，就会被删除。 （2）大小限制 根据Topic存储大小，当Topic所占日志的大小大于一个阈值，则可以开始删除最旧的消息。Kafka会启动一个新的线程，定期检查是否存在可以删除的消息。 日志压缩 很多场景中，Kafka消息的key与value值会不断变化，就像数据库中的数据会不断被修改，消费者只会关心最新的key对应的value。如果开启日志压缩功能，Kafka会开启线程，定时对相同key的消息进行合并，并保留最新的value值。 Broker 独立的Kafka服务就是一个broker，broker主要的工作就是接受生产者发送来的消息，分配offset并保存到磁盘中。Broker除了接受生产者发送的消息，还处理消费者、其他Broker的请求，根据请求类型进行相应处理行和响应返回。正常情况下一台机器对应一个broker。 副本 所谓副本就是对消息进程冗余备份，分布式系统在不同机器上相互保存对方数据。在Kafka中，每个分区（partition）可以有多个副本，每个副本中的消息是一样的（在同一时刻，多台机器之间的消息并不完全一致）。 生产者 生产者（Producer）的主要工作是生成消息。将消息发布根据规则推送到Topic的对应分区中。例如：（1）对key进行hash;（2）轮询；（3）自定义。 消费者 消费者（Consumer）的主要工作消费消息。从对应分区中拉取Topic的消息进行消费。消费者需要通过offset记录自己的消费位置。 消费者组 多个消费者（Consumer）构成消费者组（Consumer Group）。消费者组（Consumer Group）订阅的主题（Topic）的每个分区只能被分配给在同一个消费者组中的一个消费者处理。但一个消费者可以消费同一主题（Topic）的多个分区。

消息传递模式 kafka没有消息推送，只有消息拉取。但消费者可以通过轮询拉取的方式实现消息推送功能。

Kafka架构概图

五、核心特性详解

控制器选举及恢复 控制器是Kafka的核⼼组件之⼀，它的主要作⽤是在 ZooKeeper 的帮助下协调和管理整个Kafka集群。 Kafka 利⽤ZooKeeper 的领导者选举机制，每个Broker 都会参与竞选主控制器，但是最终只会有⼀个 Broker 可以成为主控制器。

控制器有以下⼏个职责：

1. 监听分区相关的变化，例如：运⾏kafka-reassign-partitions.sh 脚本对已有主题分区的细粒度的分配功能
2. 监听主题相关的变化
3. 监听broker相关的变化
4. 控制器选举：每个代理节点都会作为ZooKeeper的客户端，向ZooKeeper 服务端尝试创建 /controller 临时节点，但是最终只有 1 个Broker 可以成功创建临时节点。因为 /controller 节点是临时节点，当主控制器出现故障或者会话失效时，临时节点会被删除。此时所有的Broker 都会重新竞选 Leader，也就是尝试创建 /controller临时节点。

Kafka控制器将Broker节点信息存放在 ZooKeeper 的 /controller节点上，每个broker都会在内存中保存当前控制器的brokerId值，这个值可以标识为activeControllerId，每个broker还会对/controller节点添加监听器，以此来监听此节点的数据变化。

当/controller节点的数据发⽣变化时，每个broker都会更新⾃身内存中保存的activeControllerId。如果 broker在数据变更前是控制器，在数据变更后⾃身的brokerid值与新的activeControllerId值不⼀致，那 么就需要“退位”，关闭相应的资源。有可能控制器由于异常⽽下线，造成/controller这个临时节点被⾃动 删除；也有可能是其他原因将此节点删除了。

当/controller节点被删除时，每个broker都会进⾏选举。如果有特殊需要，则可以⼿动删除/controller节点来触发新⼀轮的选举，当然关闭控制器对应的broker以及手动向/controller节点写⼊新的brokerid所对应的数据同样可以触发新⼀轮的选举。

分区leader的选举

分区leader副本的选举由Kafka Controller 负责具体实施。当创建分区（创建主题或增加分区都有创建分区的动作）或分区上线（⽐如分区中原先的leader副本下线，此时分区需要选举⼀个新的leader上线来 对外提供服务）的时候都需要执⾏leader的选举动作。

基本思路是按照AR集合中副本的顺序查找第⼀个存活的副本，并且这个副本在ISR集合中。⼀个分区的 AR集合在分配的时候就被指定，并且只要不发⽣重分配的情况，集合内部副本的顺序是保持不变的，⽽ 分区的ISR集合中副本的顺序可能会改变。注意这⾥是根据AR的顺序⽽不是ISR的顺序进⾏选举的。举个 例⼦，集群中有3个节点:broker0、broker1、broker2，在某⼀时刻具有3个分区且副本因⼦为3的主题 quickstart的具体信息如下：

1. 此时关闭broker0，那么对于分区2⽽⾔，存活的AR就变为[1,2]，同时ISR变为[2,1]。此时查看主题 quickstart的具体信息，分区2的leader就变为了1⽽不是2。
2. 如果ISR集合中没有可⽤的副本，那么此时还需要再检查⼀下所配置的unclean.leader.election.enable参 数（默认值为false）。如果这个参数配置为true，那么表示允许从⾮ISR列表中选举leader，从AR列表 中找到第⼀个存活的副本即为leader。

当分区进⾏重分配的时候也需要执⾏leader的选举动作。这个选举策略⽐较简单：从重分配的AR列表中 找到第⼀个存活的副本，且这个副本在⽬前的ISR列表中。当发⽣优先副本的选举时，直接将优先副本设置为leader即可，AR集合中的第⼀个副本即为优先副本。

还有⼀种情况就是当某节点被优雅地关闭（也就是执⾏ControlledShutdown）时，位于这个节点上的 leader副本都会下线，所以与此对应的分区需要执⾏leader的选举。这⾥的具体思路为：从AR列表中找 到第⼀个存活的副本，且这个副本在⽬前的ISR列表中，与此同时还要确保这个副本不处于正在被关闭的 节点上。