Kafka深度解析

àì夳堔傛蜴生んèń 2022-04-25 01:52 306阅读 0赞

Kafka技术内幕笔记：

## 分区模型 ##

Kafka集群向多个消息代理服务器（brokerserver）组成，发布至Kafka集群的每条消息都有一个类别，用主题（topic）来表示。不同类型的数据，可以设置不同的主题。一个主题一般会有多个消息的订阅者，当生产者发布消息到某个主题时，订阅了这个主题的消费者都可以接收到生产者写人的新消息。Kafka集群为每个主题维护了分布式的分区（partition）日志文件，物理意义上可以把主题看作分区的日志文件（partitionedLog）。每个分区都是一个有序的、不可变的记录序列，新的消息会不断追加到提交日志（commitlog）。分区中的每条消息都会按照时间顺序分配到一个单调递增的顺序编号叫作偏移盘（offset），这个偏移量能够唯一地定位当前分区中的每一条消息。每个分区的偏移量都从0开始，不同分区之间的偏移量都是独立的不会互相影响发布到Kafka主题的每条消息包括键值和时间戳。消息到达服务端的指定分区后，都会分配到一个向增的偏移量原始的消息内容和分配的偏移量以及其他一些元数据信息最后都会存储到分区日志文件中。消息的键也可以不用设置，这种情况下消息会均衡地分布到不同的分区。

### 分区原则 ###

（1）指定了patition，则直接使用；

（2）未指定patition但指定key，通过对key的value进行hash出一个patition；

（3）patition和key都未指定，使用轮询选出一个patition。

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 </pre></td> 
 <td><pre>DefaultPartitioner类 public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
 List&lt;PartitionInfo&gt; partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) {
 int nextValue = nextValue(topic); List&lt;PartitionInfo&gt; availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() &gt; 0) {
 int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else {
 // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else {
 // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } } </pre></td> 
 </tr> 
 </tbody> 
</table>

## 消费模型 ##

消息由生产者发布到Kafka集群后会被消费者消费消息的消费模型有两种：推送模型（push)和拉取模型（pull）Kafka采用拉取模型，由消费者自己记录消费状态，每个消费者相独立地顺序读取每个分区的消息。消费者拉取的最大上限通过最高水位（watermark）控制，生产者最新写入的消息如果还没有达到备份数量，对消费者是不可见的。这种由消费者控制偏移韭量的优点是：消费者可以按照任意的顺序消费消息。比如，消费者可以重置到旧的偏移量，重新处理之前已经消费过的消息或者直接跳到最近的位置从当前时刻开始消费。在这里我们可以把分区成是一个目录,TOPIC是由PartitionLogs(分区日志)组成的

为什么要分区:为了保证读写的高效性。

## 分布式模型 ##

Kafka每个主题的多个分区日志分布式地存储在Kafka集群上，同时为了故障容错，每个分区都会副本的方式复制到多个消息代理节点上。其中一个节点会作为主副本（Leader），其他节点作为备份副本（Follower，也叫作从副本.主副本会负责所有的客户端读写操作，备份副本仅仅从主副本同步数据。当主副本现故障时，备份副本中的一个副本会被选择为新的主副本。只有主副本接受读写,每个服务端都会作为某些分区的主副本,这样Kafka集群的所有服务端整体上对客户端是负载均衡的.生产者发布消息时根据消息是否有键，采用不同的分区策略。消息没有键时，通过轮询方式进行客户端负载均衡·

分区是消费者最小并行单位增加服务器节点会提升集群的性能，增加消费者数量会提升处理性能。同一个消费组下多个消费者互相协调消费工作，Kafka会将所有的分区平均地分配给所有的消费者实例，每个消费者都可以分配到数量均等的分区，Kafka的消费组管理协议会动态地维护消费组的成员列表，当一个新消费者加入消费组，或者有消费者离开消费组，都会触发再平衡操作。

## Kafka的设计与实现 ##

预读（read-ahead）会提前将一个比较大的磁盘块读人内存。

后写（write-behind）会将很多小的逻辑写操作合并起来组合成一个大的物理写操作。并且，操作系统还会将主内存剩余的所有空闲内存空间都用作磁盘缓存（diskcache/pagecache),所有的磁盘读写操作都会经过统一的磁盘缓存（除了直接I/O会绕过磁盘缓存）。综合这几点优化特点，如果是针对磁盘的顺序访问，某些情况下它可能比随机的内存访问都要快，甚至可以和网络的速度相差无几。

消息系统内的消息从生产者保存到服务端，消费者再从服务端读取出来，数据的传输效率决定了生产者和消费者的性能。生产者如果每发送一条消息都直接通过网络发送到服务端，势必会造成过多的网络请求。如果我们能够将多条消息按照分区进行分组，并采用批量的方式一次发送一个消息集，并且对消息集进行压缩，就可以减少网络传输的带宽，进一步提高数据的传输效率。

Kafka的消息有多个订阅者的使用场景，生产者发布的消息一般会被不同的消费者消费多次。使用“零拷贝技术”（zero-copy）只需将磁盘文件的数据复制到页面缓存中一次，然后将数据从页面缓存直接发送到网络中（发送给不同的使用者时，都可以重复使用同一个页面缓存），避免了重复的复制操作。这样，消息使用的速度基本上等同于网络连接的速度了。

## Kafka生产者与消费者 ##

Kafka的生产者将消息直接发送给分区主副本所在的消息代理节点，并不需要经过任何的中间路由层。为了做到这一点，所有消息代理节点都会保存一份相同的元数据，这份元数据记录了每个主题分区对应的主副本节点。生严者客户端在发送消息之前，会向任意一个代理节点请求元数据，井确定每条消息对应的目标节点然后把消息直接发送给对应的目标节点。

生产者采用批量发送消息集的方式解决了网络请求过多的问题。生产者会尝试在内存中收集足够数据，并在一个请求中一次性发送一批数据。另外，我们还可以为生产者客户端设置“在指定的时间内收集不超过指定数量的消息”。比如，设置消息大小上限等于64字节，延迟时间等于100毫秒，表示在100毫秒内消息大小达到64字节要立即发送；如果在100毫秒时还没达到64字节，也要把已经收集的消息发送出去客户端采用这种缓冲机制，在发送消息前会收集尽可能多的数据，通过每次牺牲一点点额外的延迟来换取更高的吞吐量。相应地服务端的l/O消耗也会大大降低。

消费者读取消息、有两种方式。第一种是消息代理主动地“推送”消息、给下游的消费者，由消息代理控制数据传输的速率，但是消息代理对下游消费者是否能及时处理不得而知。如果数据的消费速率低于产生速率，消费者会处于超负荷状态，那么发送给消费者的消息就会堆积得越来越多。而且，推送方式也难以应付不同类型的消费者，因为不同消费者的消费速率不一定都相同，消息代理需要调整不同的传输速率。并让消费者充分利用系统的资源，这种方式实现实现起来比较困难。

第二种读取方式是消费者从消息代理主动地“拉取”数据，消息代理是无状态的，它不需要标记哪些消息被消费者处理过，也不需要保证一条消息只会被一个消费者处理。而且，不同的消费者可以按照向己最大的处理能力来拉取数据，即使有时候某个消费者的处理速度稍微落后，它也不会影响其他的消费者，并且在这个消费者恢复处理速度后，仍然可以追赶之前落后的数据。

因为消息系统不能作为严格意义上的数据库，所以保存在消息系统中的数据，在不用之后应该及时地删除掉并释放磁盘空间。消息需要删除，其原因一般是消息被消费之后不会再使用了，大多数消息系统会在消息代理记录关于消息是否已经被消费过的状态：当消息从消息代理发送给消费者时（基于推送模型），消息代理会在本地记录这条消息“已经被消费过了”。但如果消费者没能处理这条消息（比如由于网络原因、请求超时或消费者挂掉），就会导致“消息丢失”。解决消息丢失的一种办法是添加应答机制，消息代理在发送完消息后只把消息标记为“已发送”，只有收到消费者返回的应答信息才表示“己消费”。但还是会存在一个问题：消费者处理完消息就失败了，导致应答没有返回给消息代理，这样消息代理又会重新发送消息，导致消息被重复处理。这种方案还有一个缺点：消息代理需要保存每条消息的多种状态（比如，消息状态为“已发送”时，消息代理需要锁住这条消息，保证－消息不会发送两次），这种方式需要在客户端和服务端做一些复杂的状态一致性保证。

Kafka采用了基于拉取模型的消费状态处理，它将主题分成多个有序的分区，任何时刻每个分区都只被一个消费者使用。并且，消费者会记录每个分区的消费进度（即偏移量）。每个消费者只需要为每个分区记录一个整数值，而不需要像其他消息系统那样记录每条消息的状态。假设有10000条消息，传统方式需要记录10000条消息的状态；如果用Kafka的分区机制，假设有10个分区，每个分区1000条消息，总共只需要记录10个分区的消费状态（需要保存的状态数据少了很多，而且也没有了锁）。

和传统方式需要跟踪每条消息的应答不同，Kafka的消费者会定时地将分区的消费进度保存成检查点文件，表示“这个位置之前的消息都已经被消费过了”。传统方式需要消费者发送每条消息的应答，服务端再对应答做出不同的处理；而Kafka只需要让消费者记录消费进度，服务端不需要记录消息的任何状态。除此之外，让消费者记录分区的消费进度还有一个好处：消费者可以“故意”回退到某个旧的偏移量位置，然后重新处理数据。虽然这种处理方式看起来违反了队列模型的规定（一条消息发送给队列的一个消费者之后，就不会被其他消费者再次处理），但在实际运用中，很多消费者都需要这种功能。比如，消费者的处理逻辑代码出现了问题，在部署并启动消费者后，需要处理之前的消息并重新计算。

和生产者采用批量发送消息类似，消费者拉取消息也可以一次拉取一批消息。消费者客户端拉取消息，然后处理这一批消息，这个过程一般套在一个死循环里，表示消费者永远处于消费消息的状态（因为消息系统的消息总是一直产生数据，所以消费者也要一直消费消息）。消费者采用拉取方式消费消息有一个缺点：如果消息代理没有数据或者数据量很少，消费者可能需要不断地**轮询**，并等待新数据的到来（拉取模式主动权在消费者手里，但是消费者并不知道消息代理有没有新的数据；如果是推送模式，只有新数据产生时，消息代理才会发送数据给消费者，就不存在这种问题）。解决这个问题的方案是：允许消费者的拉取请求以阻塞式、长轮询的方式等待，直到有新的数据到来。我们可以为消费者客户端设置“指定的字节数量”，表示消息代理在还没有收集足够的数据时，客户端的拉取请求就不会立即返回。

## 副本机制和容错处理 ##

Kafka的副本机制会在多个服务端节点（简称节点即消息代理节点）上对每个主题分区的日志进行复制。当集群中的某个节点出现故障时，访问故障节点的请求会被转移到其他正常节点的副本上。副本的单位是主题的分区，Kafka每个主题的每个分区都有一个主副本以及0个或多个备份副本。备份副本会保持和主副本的数据同步，用来在主副本失效时替换为主副本。

分布式系统的容错需要确定分区是否处于存活。存活定义有两个条件：

1.节点必须和ZK保持会话；  
2.如果这个节点是某个分区的备份副本，它必须对分区主副本的写操作进行复制，并且复制的进度不能落后太多。

满足这两个条件，叫作“正在同步中”（in-sync）。如果一个备份副本挂掉、没有响应或者落后太多，主副本就会将其从同步副本集合中移除。反之，如果备份副本重新赶上主副本，它就会加入到主副本的同步集合中。在Kafka中，一条消息只有被ISR集合的所有副本都运用到本地的日志文件，才会认为消息被成功提交了。任何时刻，只要ISR至少有一个副本是存活的，Kafka就可以保证一条消息一旦被提交，就不会丢失”。只有已经提交的消息才能被消费者消费，因此消费者不用担心会看到因为主副本失败而丢失的消息。

## Kafka生产过程分析 ##

### 写入方式 ###

producer采用推（push）模式将消息发布到broker，每条消息都被追加（append）到分区（patition）中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障kafka吞吐率）。

![FvxUde.png][]

(1）producer先从zookeeper的 “/brokers/…/state”节点找到该partition的leader

(2) producer将消息发送给该leader

(3）leader将消息写入本地log

(4）followers从leader pull消息，写入本地log后向leader发送ACK

(5) leader收到所有ISR中的replication的ACK后，增加HW（high watermark，最后commit 的offset）并向producer发送ACK

### 存储策略 ###

无论消息是否被消费，kafka都会保留所有消息。有两种策略可以删除旧数据：

(1) 基于时间：log.retention.hours=168