Kafka基础知识_kafka重试次数

        引用kafka官网的一句话：Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications. (Apache Kafka 是一个开源分布式事件流平台，被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用程序)

1. Kafka 快速入门

        Kafka 是一个分布式消息引擎与流处理平台。

1.1 Kafka 术语

1. Producer：生产者，消息产生和发送端。

2. Broker：Kafka 实例，多个 broker 组成一个 Kafka 集群，通常一台机器部署一个 Kafka 实例，一个实例挂了不影响其他实例。

3. Consumer：消费者。 一个 topic 可以让若干个消费者进行消费，若干个消费者组成一个 Consumer Group 即消费组，一条消息只能被消费组中一个 Consumer 消费。

4. Topic：主题，服务端消息的逻辑存储单元。一个 topic 通常包含若干个 Partition 分区。

5. Partition：topic 的分区，分布式存储在各个 broker 中， 实现发布与订阅的负载均衡。若干个分区可以被若干个 Consumer 同时消费，达到消费者高吞吐量。一个分区拥有多个副本（Replica）,高可用。

6. message：消息，是 Kafka 服务端实际存储的数据，每一条消息都由一个 key、一个 value 以及消息时间戳 timestamp 组成。

7. offset：偏移量，分区中的消息位置，由 Kafka 自身维护，Consumer 消费时也要保存一份 offset 以维护消费过的消息位置。
8. Zookeeper： Kafka 集群能够正常工作，需要依赖于 zookeeper，zookeeper 帮助 Kafka 存储和管理集群信息。
9. Leader： 每个分区多个副本的“主”副本，生产者发送数据的对象，以及消费者消费数据的对象，都是 leader。
10. Follower： 每个分区多个副本的“从”副本，实时从 leader 中同步数据，保持和 leader 数据的同步。leader 发生故障时，某个 follower 还会成为新的 leader。

1.2 Kafka 作用与特点

• 高吞吐、低延时：Kafka 能够达到百万级的消息吞吐量，延迟可达毫秒级；

• 持久化存储：Kafka 的消息最终持久化保存在磁盘之上，提供了顺序读写以保证性能，并且通过 Kafka 的副本机制提高了数据可靠性。

• 分布式可扩展：Kafka 的数据是分布式存储在不同 broker 节点的，以 topic 组织数据并且按 partition 进行分布式存储，整体的扩展性都非常好。

• 高容错性：集群中任意一个 broker 节点宕机，Kafka 仍能对外提供服务。

1.3 Kafka 体系架构

        Kafka 体系架构包括若干 Producer、若干 Broker、若干 Consumer，以及一个 ZooKeeper 集群。(如下图)

 2. Kafka 消息发送机制

        生产端的基本流程如下图所示：

2.1 异步发送

Kafka 自从 0.8.2 版本就引入了新版本 Producer API，新版 Producer 完全是采用异步方式发送消息。生产端构建的 ProducerRecord 先是经过 keySerializer、valueSerializer 序列化后，再是经过 Partition 分区器处理，决定消息落到 topic 具体某个分区中，最后把消息发送到客户端的消息缓冲池 accumulator 中，交由一个叫作 Sender 的线程发送到 broker 端。

这里缓冲池 accumulator 的最大大小由参数 buffer.memory 控制，默认是 32M，当生产消息的速度过快导致 buffer 满了的时候，将阻塞 max.block.ms 时间，超时抛异常，所以 buffer 的大小可以根据实际的业务情况进行适当调整。

2.2 批量发送

        发送到缓冲 buffer 中消息将会被分为一个一个的 batch，分批次的发送到 broker 端，批次消息大小由参数 batch.size 控制，默认16KB。意味着正常情况下消息到了 16KB 时才会批量发送到 broker 端，所以一般减小 batch 大小有利于降低消息延时，增加 batch 大小有利于提升吞吐量。

那么生成端消息是不是必须要达到一个 batch 大小时，才会批量发送到服务端呢？答案是否定的，Kafka 生产端提供了另一个重要参数 linger.ms，该参数控制了 batch 最大的空闲时间，超过该时间的 batch 也会被发送到 broker 端。所以说这两个参数达到一个就发送。

2.3 消息重试

        Kafka 生产端支持重试机制，对于某些原因导致消息发送失败的，比如网络抖动等，开启重试后 Producer 会尝试再次发送消息。由参数 retries 控制，参数含义代表重试次数，默认值为 0 表示不重试，建议设置大于 0 比如 3。

 2. Kafka 消息消费机制 

         Kafka Consumer 采用的是主动拉取 Broker 数据进行消费的即 Pull 模式。消费端的基本流程如下：

        上图得知：一个 topic 可以让若干个消费者进行消费，若干个消费者组成一个 Consumer Group 即消费组 ，一条消息只能被消费组中的一个消费者进行消费。 所以必然会涉及到 partition 的分配问题，即确定哪个 partition 由哪个 consumer 来消费。 Kafka 有两种分配策略，一个是 RoundRobin，一个是 Range，最新还有一个StickyAssignor策略。默认为Range，当消费者组内消费者发生变化时，会触发分区分配策略。

2.1 Range(范围分区) 策略

Range是对每个Topic而言的（即一个Topic一个Topic分），首先对同一个Topic里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。然后用Partitions分区的个数除以消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽，那么前面几个消费者线程将会多消费一个分区。

不足：当消费者组内订阅的主题越多，分区分配可能越不均衡。

2.2  RoundRobin（轮询分区）策略

         RoundRobin 轮询分区策略，是把所有的 partition 和所有的 consumer 都列出来，然后按照 hascode 进行排序，最后通过轮询算法来分配 partition 给到各个消费者。

不足：当消费者组内订阅不同主题时，可能造成消费混乱。如果某个消费者没有订阅消费组内的某个 topic，那么在分配分区的时候，此消费者将不会分配到这个 topic 的任何分区。 

2.3  StickyAssignor 策略

        (后面补充...)

2.4  副本数据同步策略

（1）何时发送 ack？

确保有 follower 与 leader 同步完成，leader 再发送 ack，这样才能保证 leader 挂掉之后，能在 follower 中选举出新的 leader 而不丢数据。

（2）多少个 follower 同步完成后发送 ack？

全部 follower 同步完成，再发送 ack。

2.4.1 ISR

采用第二种方案，所有 follower 完成同步，producer 才能继续发送数据，设想有一个 follower 因为某种原因出现故障，那 leader 就要一直等到它完成同步。这个问题怎么解决？ leader维护了一个动态的 in-sync replica set（ISR）：和 leader 保持同步的 follower 集合。当 ISR 集合中的 follower 完成数据的同步之后，leader 就会给 follower 发送 ack。如果 follower 长时间未向 leader 同步数据，则该 follower 将被踢出 ISR 集合，该时间阈值由 replica.lag.time.max.ms 参数设定。leader 发生故障后，就会从 ISR 中选举出新的 leader。

2.4.2 ack 应答机制

对于某些不太重要的数据，对数据的可靠性要求不是很高，能够容忍数据的少量丢失，所以没必要等 ISR 中的 follower 全部接受成功。 所以 Kafka 为用户提供了三种可靠性级别，用户根据可靠性和延迟的要求进行权衡，选择以下的配置。

（1）ack 参数配置：

• 0：producer 不等待 broker 的 ack，这提供了最低延迟，broker 一收到数据还没有写入磁盘就已经返回，当 broker 故障时有可能丢失数据。
• 1：producer 等待 broker 的 ack，partition 的 leader 落盘成功后返回 ack，如果在 follower 同步成功之前 leader 故障，那么将会丢失数据。
• -1（all）：producer 等待 broker 的 ack，partition 的 leader 和 follower 全部落盘成功后才返回 ack。但是在 broker 发送 ack 时，leader 发生故障，则会造成数据重复。

2.4.3 故障处理细节

LEO：每个副本最大的 offset。 HW：消费者能见到的最大的 offset，ISR 队列中最小的 LEO。

（1）Follower 故障

follower 发生故障后会被临时踢出 ISR 集合，待该 follower 恢复后，follower 会 读取本地磁盘记录的上次的 HW，并将 log 文件高于 HW 的部分截取掉，从 HW 开始向 leader 进行同步数据操作。等该 follower 的 LEO 大于等于该 partition 的 HW，即 follower 追上 leader 后，就可以重新加入 ISR 了。

（2）Leader 故障

leader 发生故障后，会从 ISR 中选出一个新的 leader，之后，为保证多个副本之间的数据一致性，其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉，然后从新的 leader 同步数据。 注意：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。 5.3 Exactly Once 语义

将服务器的 ACK 级别设置为-1，可以保证 producer 到 server 之间不会丢失数据，即 At Least Once 语义。相对的，将服务器 ACK 级别设置为0，可以保证生产者每条消息只会被发送一次，即At Most Once 语义。 At Least Once 可以保证数据不丢失，但是不能保证数据不重复；相对的，At Most Once 可以保证数据不重复，但是不能保证数据不丢失。

但是，对于一些非常重要的信息，比如交易数据，下游数据消费者要求数据既不重复也不丢失，即 Exactly Once 语义。 0.11版本的 Kafka，引入了幂等性：producer 不论向 server 发送多少重复数据，server 端都只会持久化一条。即：

At Least Once + 幂等性 = Exactly Once 

要启用幂等性，只需要将 producer 的参数中 enable.idompotence 设置为 true 即可。开启幂等性的 producer 在初始化时会被分配一个 PID，发往同一 partition 的消息会附带 Sequence Number。而 borker 端会对 <PID,Partition,SeqNumber> 做缓存，当具有相同主键的消息提交时，broker 只会持久化一条。 但是 PID 重启后就会变化，同时不同的 partition 也具有不同主键，所以幂等性无法保证跨分区会话的 Exactly Once。

3. Kafka 消费端 Rebalance 机制

3.1 Rebalance 概念

        Rebalance 是让一个消费组的所有消费者就如何消费订阅 topic 的所有分区达成共识的过程，在 Rebalance 过程中，所有 Consumer 实例都会停止消费，等待 Rebalance 的完成。因为要停止消费等待重平衡完成，因此 Rebalance 会严重影响消费端的 TPS，是应当尽量避免的。

3.2 Rebalance 发生条件

关于何时会发生 Rebalance，总结起来有三种情况：

• 消费组的消费者成员数量发生变化

• 消费主题的数量发生变化

• 消费主题的分区数量发生变化

其中后两种情况一般是计划内的，比如为了提高消息吞吐量增加 topic 分区数，这些情况一般是不可避免的。

3.3 Kafka 协调器

在介绍如何避免 Rebalance 问题之前，先来认识下 Kafka 的协调器 Coordinator，和之前 Kafka 控制器类似，Coordinator 也是 Kafka 的核心组件。

主要有两类 Kafka 协调器：

• 组协调器（Group Coordinator）

• 消费者协调器（Consumer Coordinator）

Kafka 为了更好的实现消费组成员管理、位移管理，以及 Rebalance 等，broker 服务端引入了组协调器（Group Coordinator），消费端引入了消费者协调器（Consumer Coordinator）。每个 broker 启动的时候，都会创建一个 GroupCoordinator 实例，负责消费组注册、消费者成员记录、offset 等元数据操作，这里也可以看出每个 broker 都有自己的 Coordinator 组件。另外，每个 Consumer 实例化时，同时会创建一个 ConsumerCoordinator 实例，负责消费组下各个消费者和服务端组协调器之前的通信。可以用下图表示协调器原理：

客户端的消费者协调器 Consumer Coordinator 和服务端的组协调器 Group Coordinator 会通过心跳不断保持通信。

3.4 如何避免消费组 Rebalance

有新的消费者加入，通常是我们为了提高消费速度增加了消费者数量，比如增加了消费线程或者多部署了一份消费程序；有消费者退出，这种情况多是和我们消费端代码有关，是我们要重点避免的。

正常情况下，每个消费者都会定期向组协调器 Group Coordinator 发送心跳，表明自己还在存活，如果消费者不能及时的发送心跳，组协调器会认为该消费者已经“死”了，就会导致消费者离组引发 Rebalance 问题。这里涉及两个消费端参数：session.timeout.ms 和 heartbeat.interval.ms，含义分别是组协调器认为消费组存活的期限，和消费者发送心跳的时间间隔，其中 heartbeat.interval.ms 默认值是3s，session.timeout.ms 在 0.10.1 版本之前默认 30s，之后默认 10s。另外，0.10.1 版本还有两个值得注意的地方：

• 从该版本开始，Kafka 维护了单独的心跳线程，之前版本中 Kafka 是使用业务主线程发送的心跳。

• 增加了一个重要的参数 max.poll.interval.ms，表示 Consumer 两次调用 poll 方法拉取数据的最大时间间隔，默认值 5min，对于那些忙于业务逻辑处理导致超过 max.poll.interval.ms 时间的消费者将会离开消费组，此时将发生一次 Rebalance。

此外，如果 Consumer 端频繁 FullGC 也可能会导致消费端长时间停顿，从而引发 Rebalance。因此，我们总结如何避免消费组 Rebalance 问题，主要从以下几方面入手：

• 合理配置 session.timeout.ms 和 heartbeat.interval.ms，建议 0.10.1 之前适当调大 session 超时时间尽量规避 Rebalance。

• 根据实际业务调整 max.poll.interval.ms，通常建议调大避免 Rebalance，但注意 0.10.1 版本之前没有该参数。

• 监控消费端的 GC 情况，避免由于频繁 FullGC 导致线程长时间停顿引发 Rebalance。

合理调整以上参数，可以减少生产环境中 Rebalance 发生的几率，提升 Consumer 端的 TPS 和稳定性。

4. 总结

目前总结了Kafka 体系架构、Kafka 消息发送机制、副本机制，Kafka 控制器、消费端 Rebalance 机制等各方面核心原理，后续其他的一些基本命令下一篇总结一下。