Kafaka核心设计与实践原理（第一部分）

设计理念

基本概念

1.1.1 主要功能与应用场景

Kafka 作为一个分布式流式处理平台，提供了三个关键的功能：

1. 消息系统：消息顺序性保障、回湖消费功能、以及传统的消息系统功能 (系统解耦、冗余存储、流量削峰、缓冲、异步通信、扩展性、可恢复性等) 。
2. 容错的持久化记录消息流：把消息持久化到磁盘，避免了消息丢失的风险。（由此也可以把 Kafka 作为长期的数据存储系统来使用，只需要把对应的数据保留策略设置为“永久”或启用主题的日志压缩功能即可）
3. 流式处理平台： 在消息发布的时候进行处理，Kafka 提供了一个完整的流式处理类库_（如窗口、连接、变换和聚合等各类操作）_。

其主要的应用场景：

1. 消息队列 ：实时流数据管道，可靠地在系统或应用程序之间获取数据。
2. 异步、削峰： 对产生的流量进行削峰和异步处理
3. 解耦： 解耦消费方和生产方对消息的处理

1.1.2 相比于其他 MQ 的优缺点

Kafka的设计目标是高吞吐量，所以他的一些特性都和这个目标相关，如

1. Kafka 使用的是顺序 I/O，为保证顺序，Kafka强制点对点的按顺序传递消息，这意味着，一个 consumer 在分区中只有一个位置。
2. Kafka 不保存每个消息的状态，不提供以随机访问的形式更新消息的状态，而是使用 offset 划分使用和未使用的消息。
3. Kafka支持点对点的批量消息传递。
4. Kafka的消息存储使用了 OS pagecache（页缓存，page cache的大小为一页，通常为4K，用于缓存文件的逻辑内容），提高读写文件时的操作效率。

和其他消息队列的对比：

broker \ topic \ partition…关系结构:

Partition（分区） :

属于 Topic 的一部分。一个 Topic 可以有多个 Partition(即多副本机制) ，并且同一 Topic 下的 Partition 可以分布在不同的 Broker 上。
同一主题下的不同分区包含的消息是不同的，分区在存储层面可以看作一个可追加的日志（Log）文件，消息在被追加到分区日志文件的时候都会分配一个特定的偏移量(offset）。offset 是消息在分区中的唯一标识，Kafka 通过它来保证消息在分区内的顺序性，不过ofiset 并不跨越分区，也就是说，Kafka 保证的是分区有序而不是主题有序。
一个topic下，不同的 broker 上多个partition的追加写入：

tips：kafka 中 ，Producer 在发布消息时可以选择指定要发送的 Partition，或者使用默认的分区策略让 Kafka 自动选择 Partition。

多副本机制是什么?
   Kafka 为 Partition 引入了多副本（Replica）机制，来实现故障自动转移。Partition 中的多个副本之间会有一个叫做 leader 的家伙，其他副本称为 follower。我们发送的消息会被发送到 leader 副本，然后 follower 副本才能从 leader 副本中拉取消息进行同步。
多副本架构：

生产者和消费者只与 leader 副本交互。
follower 保证消息存储的安全性。当 leader 副本发生故障时会从 follower 中选举出一个 leader,但是 follower 中如果有和 leader 同步程度达不到要求的参加不了 leader 的竞选。

多分区 以及多副本机制有什么好处？

1. 一个 Topic 下的多个 Partition 可以分布在不同的 Broker 上, 可以提供更好的并发能力（类似负载均衡）。
2. Partition 的 Replica 机制, 极大地提高了消息存储的安全性, 提高了容灾能力，不过也相应的增加了所需要的存储空间。

Broker

概念：

• AR(Assigned Replicas）: 所有副本 (replicas) 统称为 AR；
• ISR （In Sync Replica）: Leader 副本保持一定同步（是指可忍受flower的滞后范围，这个范围可以通过参数进行配置）的 follower 副本, 包括 leader 副本自己；
• OSR（Out-of-Sync Replicas）:与 leader 副本同生滞后过多的副本（不包括 leader 副本）组成 OSR
• HW : Highwatermark, 俗称高水位，它表示了一个特定的消息偏移量(offset), 在一个 parttion中consumer只能拉取这个 offset 之前的消息(此 offset 跟 consumer offset 不是一个概念) ;
• LEO: LogEndOffset , 日志末端偏移量, 用来表示当前日志文件中下一条写入消息的offset；

In Sync Replica：

默认情况下，当 leader 副本发生故障时，只有在ISR 集合中的副本才有资格被选举为新的 leader。
ISR 与 HW 和LEO 也有紧密的关系。Kafka 通过ISR的同步机制及优化策略，用 HW & LEO 的方式很好地确保了数据不丢失以及吞吐率。而ISR的管理最终都会反馈到 ZK 上。

ISR相关参数：

• replica.lag.time.max.ms: 配置副本与 leader 副本之间的最大同步延迟时间。如果同步延迟超过这个时间，那么它将被移出 ISR。默认值为 10,000 毫秒（10秒）。
• replica.lag.max.messages: 配置副本与 leader 副本之间的最大消息数差异。如果消息数差异超过这个值，那么它将被移出 ISR。默认值为 4,000,000 条消息。

ISR 与 HIW 和 LEO

说明：

• 日志文件的 HIW 为6，表示消费者只能拉取到ofiset 在0 至5之间的消息，而offset 为6的消息对消费者而言是不可见的
• 分区 ISR 集合中的每个副本都会维护自身的 LEO（Log End Offset），而ISR 集合中最小的 LEO即为分区的HW，对消费者而言只能消费 HW 之前的消息。

Leader和Follower之间的消息同步过程，以及HW、LEO的变化：

1. 投递消息：

2. leader写入消息

3. follower同步消息

4. ISR完成同步消息

生产者

生产者客户端的整体架构：

Kafka有很多版本的生产者客户端：

• 于 Kafka 开源之初使用 Scala 语言编写的客户端；
• 从 Kafka 0.9.x 版本开始推出的使用 Java 语言编写的客户端；
• 在Kafka官网中， “CLIENTS” 的入口提供了一份多语言的支待列表， 其中包括常用的C、C++、 Python 、 Go等语言， 不过这些其他类语言的客户端并非由Kafka社区维护。

核心对象：

public class ProducerRecords<K, V>{
    private final String topic; //主题
    private final Integer partition; //分区号
    private final Headers headers; //消息头部
    private final K key; //键
    private final V value; //值
    private final Long timestamp;//消息的时间戳
//省略其他成员方法和构造方法
}
1
2
3
4
5
6
7
8
9

说明：

• headers 字段是消息的头部，Kafka 0.11.x版本才引入这个属性，它大多用来设定一些与应用相关的信息，如无需要也可以不用设置。
• key 是用来指定消息的键，它不仅是消息的附加信息，还可以用来计算分区号，进而可以让消息发往特定的分区。这个 key可以让消息再进行二次归类，同一个key 的消息会被划分到同一个分区中。
• 一个ProducerRecords最少需要指定 topic 和 value。

生产者实例化

1、KafkaProducer：
KafkaProducer 是 Kafka 客户端库提供的原生生产者实现，它提供了更底层的 API 来操作 Kafka 生产者。通过使用 KafkaProducer，您可以更细粒度地控制生产者的配置和行为，包括设置生产者的属性、自定义分区策略、实现自定义的消息发送等。
优点：

• 提供更灵活、更底层的控制，适合特定的定制化需求。
• 可以使用原生的 Kafka 类型和接口。

缺点：

• 使用更复杂，需要手动管理生产者的初始化、线程安全等。
• 需要编写更多的代码来处理异常和错误情况。

2、使用 KafkaTemplate：
KafkaTemplate 是 Spring Kafka 提供的高级封装，它简化了 Kafka 生产者的使用，并提供了更易于理解和使用的 API。通过使用 KafkaTemplate，您可以不必关心生产者的初始化、线程安全等细节，而是直接使用 KafkaTemplate 提供的方法来发送消息。
优点：

• 简化了 Kafka 生产者的使用，减少了样板代码，提高开发效率。
• Spring Kafka 提供了更友好的 API，支持将消息发送到指定的 topic、partition 和 key，以及实现异步发送等功能。

缺点：

• 提供的功能相对较为有限，如果需要更多的细粒度控制，可能需要借助额外的配置和扩展。

发送消息

KafkaProducer对外提供了两个 send()方法：

public Future<RecordMetadata> send(ProducerRecord<K, V> record)

public Future<RecordMetadata> send(ProducerRecord<K, V> record,Callback callback)
1
2
3

• RecordMetadata对象里包含了消息的一些元数据信息，比如当前消息的主题、分区号、分区中的偏移量(offset)、 时间戳等。如果在应用代码中需要这些信息，则需要使用 future.get(); 的方式来阻塞主流程，等待消息发送结果
• Callback是否有序？

消息自动重试

KafkaProducer中一般会发生两种类型的异常：可重试的异常和不可重试的异常。
常见的可重试异常有：NetworkException、LeaderNotAvailableException, UnknownTopicOrPart廿ionException、 NotEnoughReplicasException、NotCoordinatorException 等。重试配置：org.apache.kafka.clients.producer.ProducerConfig#RETRIES_CONFIG。

不可重试的异常，如 RecordTooLargeException（发送的消息太大） 异常，KafkaProducer对此不会进行任何重试， 直接抛出异常。

分区器

分区器的作用就是为消息分配分区。
如果消息 ProducerRecord 中指定了 partition字段， 那么就不需要分区器， 因为partition代表的就是所要发往的分区号。
如果消息ProducerRecord中没有 指定partition字段，那么就需要依赖分区器，根据key 这个字段来计算partition的值。

Kafka 中提供的默认分区器是 org.apache.kafka.clients.producer.internals.DefaultPartitioner

相关配置

#kafka配置，更多配置请参考：KafkaProperties
spring.kafka:
  #公共参数，其他的timeout.ms, request.timeout.ms, metadata.fetch.timeout.ms保持默认值
  properties:
    #指定producer在发送批量消息前等待的时间，当设置此参数后，即便没有达到批量消息的指定大小(batch-size)，到达时间后生产者也会发送批量消息到broker。默认情况下，生产者的发送消息线程只要空闲了就会发送消息，即便只有一条消息。设置这个参数后，发送线程会等待一定的时间，这样可以批量发送消息增加吞吐量，但同时也会增加延迟。
    linger.ms: 50 #默认值：0毫秒，当消息发送比较频繁时，增加一些延迟可增加吞吐量和性能。
    #这个参数指定producer在一个TCP connection可同时发送多少条消息到broker并且等待broker响应，设置此参数较高的值可以提高吞吐量，但同时也会增加内存消耗。另外，如果设置过高反而会降低吞吐量，因为批量消息效率降低。设置为1，可以保证发送到broker的顺序和调用send方法顺序一致，即便出现失败重试的情况也是如此。
    #注意：当前消息符合at-least-once，自kafka1.0.0以后，为保证消息有序以及exactly once，这个配置可适当调大为5。
    max.in.flight.requests.per.connection: 1 #默认值：5，设置为1即表示producer在connection上发送一条消息，至少要等到这条消息被broker确认收到才继续发送下一条，因此是有序的。
  producer:
      #这个参数可以是任意字符串，它是broker用来识别消息是来自哪个客户端的。在broker进行打印日志、衡量指标或者配额限制时会用到。
      clientId: ${spring.application.name} #方便kafkaserver打印日志定位请求来源
      bootstrap-servers: 127.0.0.1:8080 #kafka服务器地址，多个以逗号隔开
      #acks=0：生产者把消息发送到broker即认为成功，不等待broker的处理结果。这种方式的吞吐最高，但也是最容易丢失消息的。
      #acks=1：生产者会在该分区的leader写入消息并返回成功后，认为消息发送成功。如果群首写入消息失败，生产者会收到错误响应并进行重试。这种方式能够一定程度避免消息丢失，但如果leader宕机时该消息没有复制到其他副本，那么该消息还是会丢失。另外，如果我们使用同步方式来发送，延迟会比前一种方式大大增加（至少增加一个网络往返时间）；如果使用异步方式，应用感知不到延迟，吞吐量则会受异步正在发送中的数量限制。
      #acks=all：生产者会等待所有副本成功写入该消息，这种方式是最安全的，能够保证消息不丢失，但是延迟也是最大的。
      acks: all #默认值：1
      #当生产者发送消息收到一个可恢复异常时，会进行重试，这个参数指定了重试的次数。在实际情况中，这个参数需要结合retry.backoff.ms来使用，建议总的重试时间比集群重新选举leader的时间长，这样可以避免生产者过早结束重试导致失败。
      #另外需注意，当开启重试时，若未设置max.in.flight.requests.per.connection=1，则可能出现发往同一个分区的两批消息的顺序出错，比如，第一批发送失败了，第二批成功了，然后第一批重试成功了，此时两者的顺序就颠倒了。
      retries: 2  #发送失败时重试多少次，0=禁用重试（默认值）
      retry-backoff-ms: 1000 #重试等待间隔
      #默认情况下消息是不压缩的，此参数可指定采用何种算法压缩消息，可取值：none,snappy,gzip,lz4。snappy压缩算法由Google研发，这种算法在性能和压缩比取得比较好的平衡；相比之下，gzip消耗更多的CPU资源，但是压缩效果也是最好的。通过使用压缩，我们可以节省网络带宽和Kafka存储成本。
      compressionType: "none" #如果不开启压缩，可设置为none（默认值），比较大的消息可开启。
      #当多条消息发送到一个分区时，Producer会进行批量发送，这个参数指定了批量消息大小的上限（以字节为单位）。当批量消息达到这个大小时，Producer会一起发送到broker；但即使没有达到这个大小，生产者也会有定时机制来发送消息，避免消息延迟过大。
      batch-size: 16384 #默认16K，值越小延迟越低，但是吞吐量和性能会降低。0=禁用批量发送
      #这个参数设置Producer暂存待发送消息的缓冲区内存的大小，如果应用调用send方法的速度大于Producer发送的速度，那么调用会阻塞一定（max.block.ms）时间后抛出异常。
      buffer-memory: 33554432 #缓冲区默认大小32M


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

专题讨论

如何保证消息的可靠性（或如何避免消息丢失）？

一条消息的流转，会经过 Producer / Broker / Consumer ，所以消息的丢失，可能出现在这三个位置，我们分情况讨论。具体的实现，可参考：消息可靠性方案的实现

Producer 端消息丢失

1. Producer 是否等待 TopicLeader 的消息回执（ack：默认为1）

acks = 1： Leader 写入消息到本地日志，则请求被认为成功。如果此时 Leader 应答请求后挂掉了，消息会丢失。

acks = 0： Producer 执行发送请求后立即返回，不等待 Leader 的确认，消息会丢失。

acks = -1：分区 Leader 必须等待消息被成功写入到所有的ISR副本(In-Sync Replace)中才认为请求成功。该种情况下配合 Producer 的消息重试机制 来保证消息可靠性。

2. Producer 进程意外停止导致的消息丢失

为了提升发送效率，减少IO次数，Producer 在发送数据时，可以将多个消息缓存在本地 buffer 中进行合并后发送（主要通过在 KafkaProducer 对象中配置  buffer.memory 、batch.size 、linger.ms 参数来控制）。

可以看到存在本地 Buffer 的消息，在 Producer 进程突然结束后，会出现消息丢失的情况。如果有必要，可以适当调小上面提及的参数，使发送频率变高，减少丢失数据。

3. Producer 网络问题导致的消息丢失

由于 Producer 的网络问题，在经过 retries 和 retry.backoff.ms 后，依然发送失败，那么将使用其他的方案进行重试发送。

Broker 消息丢失

1. Leader 切换

Broker 的消息丢失，最容易出现在 Partition Leader 切换时，我们可以通过如下的一些参数来保证 Broker 消息的可靠性：

1. replication.factor 设置 topic 的副本
2. 服务端设置 min.insync.replicas（ISR）>1，确保切换 Leader 时有同步的 Flower。需要 replication.factor > min.insync.replicas
3. producer 端设置 acks=all，保证消息写入所有 ISR 列表中
4. producer 端设置 retries 和 retry.backoff.ms 对 Broker 存储失败的消息进行重发

Consumer 消息丢失

1. Consumer 按时间间隔自动提交 offset

使用 enable.auto.commit 和 auto.commit.interval.ms 控制异步 offset 提交，可能会出现消息丢失或者重复消费。

2. Consumer 手动提交 offset

手动提交 offset 并不能 100% 保证消息一定不丢失，(比如，消费某几条消息失败后，导致 offset 没有提交，那么默认情况下，消费者会进行重试，如果重试还是失败，消费者将继续消费下一批消息，如果后续的消费成功提交 offset ，那么消费失败的消息将会丢失)，需要配合异常重试、死信队列、死信队列消息处理机制，才能保证 Consumer 100% 不丢失消息。

如何保证消息的唯一性

消息不唯一，可能会出现在 Producer 端多次发送，在 Consumer 端多次消费，下面分情况讨论：

1. Producer 客户端通过 Producer ID（PID）和 Sequence Number 保证消息唯一

0.11.0 后的版本，引入了 Producer ID（PID）和 Sequence Number 实现 Producer 的幂等，保证一个消息只被投递一次。

2. Consumer 添加持久化的去重判断

Consumer 消息的消费和业务紧密相关，所以需建立持久化的唯一标识，来避免消息重复消费。

如何保证消息的消费顺序？

Kafka 只能为我们保证每个 Partition(分区) 中的消息有序, 同一个 Topic 下的多个 Partition 内的消息在消费时,不保证有序性.

为了保证消息的顺序消费, 有如下几种方案:

1. 1 个 Topic 只配置一个 Partition
2. Producer 发送消息的时候指定 key 和 Partition, 保证需要顺序消费的消息在同一个 Partition 中, 此时无论是哪一个 Consumer 来 pull 消息都是有序的.
   • 1 和 2从本质上来讲, 都是利用 Partition 中的有序性来实现顺序消费
3. 在消息中加入状态(如订单的预下单>支付中>支付完成>撤销), Consumer 消费消息时, 判断状态决定消费还是抛出异常, 异常后等待被重新消费
   • 通过业务来控制有序的消费消息