分区日志、日志段、日志段索引、日志段时间索引、日志段位移索引、稀疏索引文件

消息设计

JMM要求Java对象必须按照8字节对齐，未对齐部分会填充空白字节进行补齐，该操作称为padding。
Kafka使用Java NIO的ByteBuffer来保存消息，同时依赖文件系统的页缓存机制，而非Java的堆缓存。ByteBuffer是紧凑的二进制字节结构，不需要padding操作。
Kafka消息格式有3个版本：V0、V1和V2。
消息层次分为两层：消息集合（message set）和消息。
一个消息集合包含若干个日志项，日志项封装实际的消息和一组元数据信息。Kafka日志文件是一系列消息集合日志项构成的。Kafka不会在消息层面上直接操作，而是在消息集合上进行写入。
V2之前的版本使用的是日志项（log entry），V2版本则使用消息批次（record batch）。
V2版本借鉴了Google ProtoBuffer中的Zig-zag编码方式。
PID、producer epoch和序列号等信息都是0.11.0.0版本为了实现幂等性producer和支持事务而引入的。PID标识一个幂等性producer的ID值，producer epoch标识某个PID携带的当前版本号，broker使用PID和epoch来确定当前合法的producer实例，并以此阻止过期producer向broker生产消息。序列号主要实现消息生产的幂等性。Kafka依靠序列号来辨别消息是否已经提交成功。

副本与ISR设计

一个Kafka分区本质上是一个备份日志，即利用多份相同的备份共同提供冗余机制来保证系统高可用性。这些备份被称为副本（replica）。Kafka把分区的所有副本均匀的分配到所有broker上，并从中选一个作为leader副本对外提供服务；其余的作为follow副本，被动的向leader请求数据，从而保持与leader副本的同步。
ISR，是Kafka集群动态维护的同步副本集合（in-sync replicas）。每个topic分区都有自己的ISR列表，ISR中的所有副本都与leader保持同步状态。只有ISR中的副本才能被选举为leader。
producer写入的一条消息只有被ISR中的所有副本都接收到，才被视为“已提交”状态。
0.9.0.0版本之前，参数replica.lag.max.messages用于控制follower副本落后leader副本的消息数。超过这个消息数，则该follower被视为“不同步”状态，从而被踢出ISR。
下列原因导致folllower与leader不同步：

• 请求速度追不上。
• 进程卡住。
• 新创建的副本。

在0.9.0.0版本之前，提供参数replica.lag.time.max.ms用于检测后面两种情况。前面的参数只检测第一种情况。
自0.9.0.0版本之后，Kafka去掉了replica.lag.max.messages参数，改用replica.lag.time.max.ms参数同时检测由于慢以及进程卡壳而导致的滞后（lagging）---follower副本落后leader副本的时间间隔。默认是10秒。对于“请求速度追不上”的情况，检测机制也发生了变化---如果一个follower副本落后leader的时间持续性的超过这个参数值，则该follower副本被认为是“不同步”的。
在Kafka中，水印表示的是位置信息，即位移（offset）。
任意时刻，HW指向的是实实在在的消息位置，而LEO指向下一条待写入的消息位置。消费者无法消费分区leader副本上位移大于分区HW的消息。分区HW就是leader副本的HW值。
Kafka使用HW值来决定副本备份的进度，而HW值的更新需要另一轮FETCH请求才能完成，该设计会导致下面问题：

• 备份数据丢失。
• 备份数据不一致。

Kafka在0.11.0.0版本中引入leader epoch值解决上面的基于水印备份机制的两个问题。
造成上面两个问题的根本原因在于HW值被用于衡量副本备份的成功与否，以及在出现崩溃时作为日志截断的依据。但HW值的更新是异步延迟，特别是需要额外的FETCH请求流程才能更新，故这中间发生任何崩溃都可能导致HW值的过期。
leader端用一段内存区域保持leader的epoch消息。所谓领导者epoch，实际是一对值（epoch,offset）。epoch表示leader的版本号，从0开始，当leader变更时，epoch自动加1，offset对应该epoch版本的leader第一条消息的位移。
每个副本引入新的状态来保持自己当leader时开始写入的第一条消息的offset以及leader版本，这样恢复的时候使用这些信息而非水位来判断是否需要截断日志。

日志存储设计

Kafka不会直接将原生消息写入日志文件，而是将消息和必要的元数据信息打包在一起封装成一个record写入日志。也就是消息集合或消息batch。
Kafka自定义了消息格式并且在写入日志前序列化成紧凑的二进制字节数组来保存日志。
Kafka的日志是以分区为单位，每个分区都有自己的日志，该日志被称为分区日志（partition log）。具体对每个日志而言，细分为日志段文件（log segment file）以及日志段索引文件。
日志段索引文件包括位移索引文件（.index）和时间索引文件（.timeindex）。它们都属于稀疏索引文件（sparse index file），Kafka不会为每条消息保存对应的索引项。而是写入若干条记录后才增加一个索引项。由broker端的log.index.interval.bytes参数控制。默认值是4KB。索引文件默认是10MB。
通过时间索引查找后，还需要到位移索引文件中定位真正的物理文件位置。

底层文件系统

Kafka使用日志段文件的第一条记录对应的offset来命名.log文件。默认大小是1GB。日志段文件填满后，会自动创建一组新的日志段文件和索引文件---这被称为日志切分（log rolling）。

• 日志留存。默认清除7天前的日志信息。或者基于日志文件大小（默认不清除）。
• 日志压实。确保topic每个分区下的每个相同key的消息都保存最新value。日志压实是topic级别的。

clients端请求处理流程

• 确认目标broker。
• 创建与broker的TCP连接并一直保持

JAVA版本clients采有了类似于Linux select和epoll的实现机制，在底层把I/O操作完全托管给JAVA NIO的Selector，故在轮询这一步中clients会检查有无真正的I/O事件发生。比如发送请求或获取请求甚至是连接重建或断开。若不是连接断开，clients会对接收到的响应指向对应的回调逻辑；如果是连接断开，则clients会自动重建连接。

broker端请求处理流程

broker启动时会创建一个请求阻塞队列，用于接收从clients端发送的请求。创建若干个请求处理线程来获取并处理该阻塞队列中的请求。

controller设计

在Kafka集群中，某个broker作为控制器。用于管理集群中所有分区的状态并执行相应的管理操作。
controller维护两类状态：每台broker上的分区副本和每个分区的leader副本信息。从维度上看，这些状态分为副本状态和分区状态。

broker请求处理流程

Kafka broker处理请求的模式是reactor设计模式。reactor设计模式是一种事件处理模式，旨在处理多个输入源同时发送过来的请求。reactor模式中的服务请求器或分发器将入站请求按照多路复用的方式分发到对应的请求处理器。
在Kafka中对应于reactor模式的“事件”实际上是连向broker的Socket连接通道，而不是clients端发送过来的真实请求。
多路复用（multiplexing），用很少的线程维护多个连接。clients端通常会保存与broker的长连接，因此不需要频繁的重建socket连接，故broker端固定使用一个acceptor线程来唯一的监听入站连接。
目前，broker以线程组而非线程池的方式来实现process线程组，之后使用数组索引轮询方式依次给每个process线程分配任务，实现了负载均衡。
Kafka创建KafkaRequestHandler线程池专门处理真正的请求。process只是将socket连接上接收到的请求放入请求队列中。这个请求队列默认大小是500，超过后会阻塞。
broker还会创建与process线程数等量的响应队列。
Kafka在设计上使用了JAVA NIO的Selector + Channel + Buffer的思想，在每个process线程中维护一个Selector实例，并通过这个Selector管理多个通道上的数据交互，这就是多路复用在process线程上的应用。

实现精确一次处理语义

幂等性producer（idempotent producer）

若一个操作执行多次的结果与只执行一次的结果是相同的，那么称该操作为幂等操作。
如果要启用幂等性producer以及获取其提供的EOS语义，需要设置producer端的参数enable.idempotent为true。
幂等性producer的设计思路类似于TCP的工作方式。发送到broker端的每批消息都会被赋予一个序列号用于消息去重。与TCP不同的是，这个序列号不会被丢弃，而是会被保存在底层日志中，这样即使分区的leader副本挂掉，新选出来的leader broker也能执行消息去重。
除了序列号，Kafka还为每个producer实例分配一个producer id（PID）。
对于PID、分区和序列号的关系，设想一个Map，key是（PID，分区号），value就是序列号。即每对（PID，分区号）都有对应的序列号值。若发送消息的序列号小于或等于broker端保存的序列号，那么broker会拒绝这条消息的写入。
由于每个新的producer实例都会被分配不同的PID。当前设计只能保证单个producer实例的EOS语义，而无法实现多个producer实例一起提供EOS语义。

事务

对事务的支持是Kafka实现EOS的第二个利器。事务使得clients端程序能够将一组消息放入一个原子性单元中统一处理。

Kakfa Broker Leader的选举

Kakfa Broker集群受zk管理。broker节点去zk注册一个临时节点，因为只有一个broker会注册成功，这个成功注册临时节点的Broker会成为Broker Controller，其他的broker角色是follower。（这个过程叫Controller在ZooKeeper注册Watch）。
这个Controller会监听其他的Broker的所有信息，如果broker controller宕机了，在zookeeper上面的那个临时节点就会消失，此时所有的broker又会一起去zk上注册一个临时节点，因为只有一个broker会注册成功。
例如：一旦有某个broker宕机，broker controller会读取该宕机broker上所有的partition在zookeeper上的状态，并选取ISR列表中的一个replica作为partition leader（如果ISR列表中的replica全挂，选一个幸存的replica作为leader; 如果该partition的所有的replica都宕机了，则将新的leader设置为-1，等待恢复，等待ISR中的任一个replica“活”过来，并且选它作为leader；或选择第一个“活”过来的Replica（不一定是ISR中的）作为Leader），这个broker宕机的事情，broker controller也会通知zk，zk就会通知其他的kafka broker。

Broker Leader的任务
（1）传送消息：producer先把message发送到partition leader，再由leader发送给其他partition follower。（如果让producer发送给每个replica那就太慢了）
（2）在向Producer发送ACK前需要保证有多少个Replica已经收到该消息：根据ack配的个数而定。
（3）处理某个Replica不工作的情况：如果这个不工作的partition replica不在ack列表中，就是producer在发送消息到partition leader上，partition leader向partition follower发送message没有响应而已，这个不会影响整个系统，也不会有什么问题。如果这个不工作的partition replica在ack列表中的话，producer发送的message的时候会等待这个不工作的partition replca写message成功，但是会等到time out，然后返回失败因为某个ack列表中的partition replica没有响应，此时kafka会自动的把这个不工作的partition replica从ack列表中移除，以后的producer发送message的时候就不会有这个ack列表下的这个部工作的partition replica了。
（4）处理Failed Replica恢复回来的情况：如果partition replica之前在ack列表中，重启后，需要把这个partition replica手动加到ack列表中。（ack列表是手动添加的，出现某个不工作的partition replica的时候自动从ack列表中移除的）

Partition leader与follower

partition也有leader和follower之分。leader是主partition，producer写kafka的时候先写partition leader，再由partition leader push给其他的partition follower。partition leader与follower的信息受zk控制，一旦partition leader所在的broker节点宕机，zookeeper会从其他的broker的partition follower上选择follower变为parition leader。

集群管理

ZK临时节点的生命周期和客户端会话绑定，从而用ZK临时节点来管理broker生命周期。

通信协议

Kafka的通信协议是基于TCP的二进制协议。

跨机房备份

MirroMaker可以实现在两个Kafka集群间拷贝数据。从而实现跨机房的数据传输。

参考

《Apache Kafka实战》
https://www.jianshu.com/p/3d2bbbeea14f