本栏目讲解kafka相关的知识，包括简介、集群部署、架构及spring整合

---

---

组成元素

• producer：消息生产者，即向 broker 发送消息的客户端
• consumer：消息消费者，即向 broker 获取消息的客户端
• consumer group：消费者组，一个消费者组由多个消费者组成，且每个消费者负责消费不同分区的数据
• broker：kafka 服务器。一个集群由多个 broker 组成，一个 broker 可以容纳多个主题
• topic：主题，即消息的分类，一个主题可以分布到多个 broker 上，一个主题分为多个分区
• partition：分区，用来存储消息。一个主题下的多个分区可以分布在多个 broker 上，提高负载和并发
• leader：主服务器，即生产者发送数据的对象，以及消费者消费数据的对象都是 leader
• follower：从服务器，实时从主服务器中同步数据，也就是主服务器的副本，当主服务器发生故障时，某个从服务器会成为新的主服务器
• zookeeper：注册中心，负责管理集群 broker 的上下线，所有 topic 的分区副本分配和 leader 选举等工作

---

工作流程

• 存储消息：生产者客户端将消息发送给 broker，broker 将数据存储到 partition 中， patition 由多个 segment 组成，segment 由 log 和 index 文件组成，log 文件用来存放实际数据，index 文件用来存放索引及数据偏移量。broker 的副本自动进行数据同步
• 消费消息：消费者客户端向 broker 发送获取数据请求，broker 根据 topic、偏移量等参数，先通过二分查找法在 index 文件中找到数据对应的起始偏移量，再根据偏移量在 log 文件中找到对应的数据返回给客户端

---

生产者相关操作

1、分区策略

• 在明确 partition 值的情况下，直接将消息存储到该 parition
• 指明 key 值，但未明确 partition 值的情况下，将 key 的 hash值 % partition总数得到 partition
• 在既没有明确 partition 值又没有明确 key 值的情况下，在第一次调用时，随机生成一个整数（后面的每次调用会在这个整数上自增），将这个随机数 % partition 总数得到 partition 值，了就是常说的轮询算法

2、数据可靠性

• 概述：将存储时间响应比较快的 follwer 存储在 ISR（同步集合在leader 发生故障之后，就会从 ISR 中选举新的 leader）中，当 ISR 的 follower 完成数据的同步之后，leader 就会发送 ack 给 producer
• ACK机制
  • 0：不等待 broker 的 ack，producer broker 一接收到还没有写入磁盘就返回，但是当 broker 故障时可能会丢失数据
  • 1：producer 等待 broker 的 ack，partition 的 leader 落盘成功后返回 ack，如果在follower同步成功之前 leader 故障，那么将会丢失数据
  • -1：producer 等待 broker 的 ack，partition 的 leader 和 follower 全部落盘成功后才返回 ack，但是如果在 follower 同步完成后，broker 发送 ack 之前，leader 发生故障，那么会造成数据重复

3、数据一致性

• follower 故障:：follower 发生故障后会被临时踢出 ISR，待该 follower 恢复后，follower 会读取本地磁盘记录的上次的 HW，并将 log 文件高于 HW 的部分截取掉，从 HW 开始向 leader 进行同步。等该 follower 的 LEO 大于等于该 partition 的 HW，就可以重新加入ISR 了
• leader 故障：leader 发生故障之后，会从 ISR 中选出一个新的 leader，为保证多个副本之间的数据一致性，其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉，然后从新的 leader 同步数据

4、Exactly Once方式

• 概述：用来保证数据既不重复也不丢失
• 原理：幂等性结合 At Least Once（acks=-1）语义等于 Exactly Once。At Least Once保证了数据的可靠性，而幂等性保证数据不重复，即将原来下游需要做的去重放在了数据上游。开启幂等性的 producer 在初始化的时候会被分配一个 PID，发往同一 partition 的消息会附带 sequenceNumber。而 broker 端会对<PID,partition,seqNumber>做缓存，当具有相同主键的消息提交时，Broker 只会持久化一条
• 启动幂等：enable.idompotence = true

---

消费者相关操作

1、消费方式

• ：使用PULL模式根据 consumer 的消费能力以适当的速率消费消息，如果当前没有数据可供消费，consumer 会等待一段时间之后再返回，避免 kafka 没有数据，消费者可能会陷入循环中，一直返回空数据

2、消费组分区策略

• 范围：以主题划分分区，用主题分区数 % 订阅主题的消费者数，得到每个消费者分配到的分区，但会出现分配不均匀的问题
• 轮询：以消费者组来划分分区，适合消费者组订阅多个相同主题的场景，即将多个主题的分区当成一个整体先进行排序，再轮询分配，但当消费者个数发生变化时，会重新轮询

3、偏移量存储

• 原因：由于 consumer 在消费过程中可能会出现断电宕机等故障，当 consumer 恢复后，需要从故障前的位置继续消费，所以 kafka 将 offset 保存在一个内置的 topic 中，该 topic 为 __consumer_offsets
• 更新方式
  • 自动提交：设置 enable.auto.commit = true，更新的频率根据参数 auto.commit.interval.ms 来定。拉取到消息后，无论成功以否都会更新 offset
  • 手动提交：设置 enable.auto.commit = false，拉取到消息后，等消费完成再调用方法 consumer.commitSync()，手动更新 offset。如果消费失败，则 offset 也不会更新，此条消息会被重复消费一次。

---

事务

• 引入原因：由于幂等性只能保证单分区单会话的不重复性，但当 broker 宕机重启后其为 producter 重新分配另一个 PID，导致之前消费的数据重复消费的可能
• 概述：producter 生成一个全局的唯一 TransactionID 来实现跨分区的会话事务，第一次拉取消息时会将 TransactionID 与 PID 进行绑定和缓存，当 producter 重启了，则会根据 TransactionID 来获得 PID