第一章 初识Kafka

Kafka定位为一个分布式流处理平台，它以高吞吐、可持久化、可水平扩展、支持流数据处理等多种特性而被广泛使用。

一个典型的Kafka体系架构包含若干Producer、若干Broker、若干Consumer，以及一个Zookeeper集群。

Zookeeper用来负责集群元数据的管理、控制器的选举等操作的，Producer将消息发送到Broker，Broker负责将收到的消息持久化到磁盘中，而Consumer负责从Broker订阅并消费消息。

在Kafka中还有两个特别重要的概念---主题(Topic)和分区(Partition)。Kafka消息以Topic为单位进行归类，Producer负责将消息发送到特定的Topic，而消费者负责订阅Topic并进行消费。

Topic是一个逻辑上的概念，它细分为多个Partition，一个Partition只属于单个Topic，同一Topic下不同Partition包含的消息是不同的，Partition在存储层可以看作是一个可追加的日志文件，消息被追加到Partition日志文件的时候会分配一个特定的偏移量(offset)。offset是消息在Partition上的唯一标识，Kafka通过它来保证消息在Partition内的顺序性，注意的是，Kafka只能保证Partition有序而不是Topic有序。(如果保证Topic有序只能让Topic包含一个Partition)。

每一个消息被发送到Broker之前，会根据Partition规则选择存储到哪个具体的Partition。如果Partition规则设定得合理，所有的消息都可以均匀地分配到不同的Partition里，如果一个Topic只对应一个文件，那么这个文件所在的机器I/O将会成为这个Topic的性能瓶颈，而Partition解决了这个问题。

Kafka为Partition引入了多副本机制，通过增加副本数量可以提升容灾能力。同一Partition的不同副本中保存的是相同的消息，副本之间是“一主多从”的关系，其中leader副本负责读写请求，follower副本只负责与leader副本的消息同步，很多时候follower副本中的消息相对leader副本有一定的滞后。副本处于不同的Broker中，当leader副本出现故障时，从follower副本中重新选举一个新的leader副本对外提供服务。Kafka通过多副本机制实现了故障的自动转移，当Kafka集群中某个Broker失效时仍然能保证服务可用。

Kafka消费端也具备一定的容灾能力。Consumer使用拉模式从服务端拉取消息，并且保存消息的具体位置，当消费者宕机后恢复上线时可以根据之前保存的消息位置重新拉取需要的消息进行消费，这样就不会造成消息丢失。

Partition中的所有副本统称为AR(Assigned Replicas)。所有与leader副本保持一定程度同步的副本(包括leader副本)组成ISR(In-Sync Replicas)，ISR时AR的一个子集。与leader副本同步滞后过多的副本组成OSR(Out-Of-Sync Replicas)。正常情况下，应该AR=ISR，OSR=0。

leader副本负责维护和跟踪ISR集合中所有follower副本的滞后状态，当follower副本落后太多或失效时，leader副本会把它从ISR集合中剔除，如果OSR中有follower副本追上了leader副本，那么leader副本会把它从OSR集合转移到ISR集合。

默认情况下，当leader副本发生故障时，只有ISR集合里的副本才有资格被选举为新的leader。

ISR与HW和LEO也有紧密的关系。HW即High Watermark的缩写，俗称高水位，标识了一个特定的消息偏移量，消费者只能拉取到这个偏移量之前的消息。LEO是Log End Offset的缩写，标识当前日志文件中下一条待写入消息的偏移量，LEO的大小相当于当前日志Partition中最后一条消息的偏移量的值加1.ISR集合中的每个副本都会维护自身的LEO，而ISR集合中最小的LEO即为Partition的HW，对消费者而言只能消费HW之前的消息。

第二章 生产者

生产者客户端的整体架构如下所示：

整个生产者有两个线程协调运行，这两个线程分别为主线程和Sender线程。主线程中由KafkaProducer创建消息，然后 通过可能的拦截器、序列化器和分区器的作用滞后缓存到消息累加器(RecordAccumulator)。Sender线程负责从消息累加器中获取消息并将其发送到Kafka中。

消息累加器主要用来缓存消息以便Sender可以批量发送，进而减少网络传输的资源消耗以提升性能。如果生产者发送消息的速度超过发送到服务器的速度，则会导致生产者空间不足，这个时候KafkaProducer的send发送调用要么被阻塞，要么抛出异常，这个取决于参数max.block.ms的配置(默认为60s)。

在消息累加器的内部为每个Partition都维护了一个双端队列，队列中的内容是ProducerBatch。消息写入缓存时，追加到双端队列的尾部。Sender从双端队列的头部读取消息。ProducerBatch是消息批次，包含多条ProducerRecord。

消息在网络上都是以字节的形式传输的，在发送之前需要创建一块内存区域来保存对应的消息。当一条消息流入消息累加器时，会先寻找与消息分区对应的双端队列，再从这个双端队列的尾部获取一个ProducerBatch，查看ProducerBatch中是否还可以写入这个ProducerRecord，可以则写入，否则创建一个新的ProducerBatch。

Sender从消息累加器中获取缓存的消息之后，会将<Partition, Deque<ProducerBatch>>的形式转变成<Node, List<ProducerBatch>的形式，Node表示Kafka集群的Broker节点，对于网络连接来说，生产者客户端与具体的Broker节点建立连接，而不关心消息属于哪个Partition，所以在这里需要一个应用逻辑层面到网络I/O层面的转换。

转换成<Node, List<ProducerBatch>的形式后，Sender还会进一步封装成<Node, Request>的形式，这样就可以将Request发送到各个Node了。Request在从Sender发送到Kafka之前还会保存到InFlightRequest中，InFlightRequest保存对象的具体形式为Map<NodeId, Deque<Request>>，它的主要作用时缓存了已经发出去但还没收到响应的请求。