前言

Kafka 是最初由 Linkedin 公司开发，是一个分布式（多节点）、支持多分区、多副本的，基于 zookeeper 协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据。

kafka 特性

1. 高吞吐量、低延迟：kafka 每秒可以处理几十万条消息，它的延迟最低只有几毫秒。

2. 可扩展性：kafka 支持热扩展，水平扩展。

3. 持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失。

4. 容错性：允许集群中节点失败。

5. 高并发：支持数千个客户端同时读写。

kafka 使用场景

• 日志收集
• 消息系统
• 监控
• 运营指标
• 流式处理
• 事件源

通信

Kafka的Producer、Broker和Consumer之间采用的是一套自行设计的基于TCP层的协议。以NIO作为网络通信的基础,把多条Message放在一起做压缩，提高压缩比率，从而在网络上传输的数据量会少一些。

主要结构

1. Broker

   kafka 集群节点,也可称为代理，在集群中每个 broker 都有一个唯一 brokerid，不得重复。如果我们想扩展kafka集群，只需引入新节点，分配一个不同的broker.id即可。

2. Topic

   • 在 Kafka 中的每一条消息都有一个 Topic。一般来说在我们应用中产生不同类型的数据，都可以设置不同的主题。
   • 一个主题一般会有多个消息的订阅者，当生产者发布消息到某个主题时，订阅了这个主题的消费者都可以接收到生产者写入的新消息。

3. Partition

   kafka 是面对分布式系统的，同时一个 topic 对应 partition，一个 partition 有对个副本,会从所有的副本中选取一个 leader 出来。所有读写操作都是通过 leader 来进行的。一个 partition 是个有序队列。

4. offset

   在分区中的每条消息都会按照时间顺序分配到一个单调递增的顺序编号，也就是我们的 Offset。Offset 是一个 Long 型的数字。

5. Producer

   负责发布消息到 Kafka broker。

6. Consumer

   消息消费者，向 Kafka broker 读取消息的客户端。

7. Consumer Group

   各个 consumer（consumer 线程）可以组成一个组（Consumer group），partition 中的每个 message 只能被组（Consumer group）中的一个 consumer（consumer 线程）消费。

kafka消费模型

消息由生产者发送到 Kafka 集群后，会被消费者消费。一般来说我们的消费模型有两种：推送模型（push）、拉取模型（pull）

1. 推送模型

   基于推送模型的消息系统，由消息代理记录消费状态。消息代理将消息推送到消费者后，标记这条消息为已经被消费，但是这种方式无法很好地保证消费的处理语义。

   比如当我们已经把消息发送给消费者之后，由于消费进程挂掉或者由于网络原因没有收到这条消息，如果我们在消费代理将其标记为已消费，这个消息就永久丢失了。

   如果我们利用生产者收到消息后回复这种方法，消息代理需要记录消费状态，这种不可取。
   如果采用 Push，消息消费的速率就完全由消费代理控制，一旦消费者发生阻塞，就会出现问题。

2. 拉取模型

   Kafka 采取拉取模型(Pull)，由自己控制消费速度，以及消费的进度，消费者可以按照任意的偏移量进行消费。比如消费者可以消费已经消费过的消息进行重新处理，或者消费最近的消息等等。

关联zookeeper

kafka 用 zk 做 meta 信息存储。

低版本时主要储存consumer 的消费状态，group 的管理以及 offset 的值。以及配合broker的选举。

后来考虑到 zk 本身的一些因素以及整个架构较大概率存在单点问题，且zk的读取效率低下，新版本中确实逐渐弱化了 zookeeper 的作用。新的 consumer 使用了 kafka 内部的 group coordination 协议，也减少了对 zookeeper 的依赖。

控制器

在Kafka集群中会有一个或者多个broker，其中有一个broker会被选举为控制器（Kafka Controller），它负责管理整个集群中所有分区和副本的状态。当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。

控制器选举

Kafka中的控制器选举工作依赖于Zookeeper，成功竞选成为控制器的broker会在Zookeeper中创建/controller临时（Ephemeral）节点，此临时节点的内容参考如下：

{“version”:1,”brokerid”:0,”timestamp”:”1593330804078”}

每个broker启动的时候会去尝试读取/controller节点的brokerid的值，如果读取到的brokerid的值不为-1，表示已经有其他broker节点成功竞选为控制器，所以当前broker就会放弃竞选；如果Zookeeper中不存在/controller节点，或者这个节点的数据异常，那么就会尝试去创建/controller节点。

Zookeeper中还有一个与控制器有关的/controller_epoch节点，这个节点是持久（Persistent）节点，节点中存放的是一个整型的controller_epoch值。controller_epoch值用于记录控制器发生变更的次数，即记录当前的控制器是第几代控制器，我们也可以称之为“控制器纪元”。

每个和控制器交互的请求都会携带controller_epoch这个字段，如果请求的controller_epoch值小于内存中的controller_epoch值，则认为这个请求是向已经过期的控制器发送的请求，那么这个请求会被认定为无效的请求。

分区副本

kafka有三层形式，kafka有多个主题，每个主题有多个分区，每个分区有多个副本。每个分区在不同的broker节点上，分区的副本又会分布在不同的broker节点上。

1. 分区个数选择

   分区的好处显而易见，提高了消息的可靠性；那是不是分区越多越好那？

   分区越多，所需要消耗的资源就越多。

   比较无脑的确定分区数的方式就是broker机器数量的2~3倍。有时间可以自己测。。。

2. follower副本和leader副本

   副本的目的就是冗余备份，且仅仅是冗余备份，所有的读写请求都是由leader副本进行处理的。follower副本仅有一个功能，那就是从leader副本拉取消息，尽量让自己跟leader副本的内容一致。

   所以也就分为leader副本与follower副本。

3. LEO和HW

   Kafka所有副本都有的两个重要属性：LEO和HW：

   • LEO：即日志末端位移(log end offset)，记录了该副本底层日志(log)中下一条消息的位移值。注意是下一条消息！也就是说，如果LEO=10，那么表示该副本保存了10条消息，位移值范围是[0, 9]。
   • High Watermark（高水位线）以下简称HW，表示消息被leader和ISR内的follower都确认commit写入本地log，所以在HW位置以下的消息都可以被消费（不会丢失）。

副本选举

zookeeper每次leader节点挂掉时，都会通过内置id，来选举处理了最新事务的那个follower节点。

实际上它是通过ISR副本集合实现,kafka会将与leader副本保持同步的副本放到ISR副本集合中。当然，leader副本是一直存在于ISR副本集合中的，在某些特殊情况下，ISR副本中甚至只有leader一个副本。

当leader挂掉时，kakfa通过zookeeper感知到这一情况，在ISR副本中选取新的副本成为leader，对外提供服务。

同步的标准是什么？跟一个参数有关：replica.lag.time.max.ms。

前面说到follower副本的任务，就是从leader副本拉取消息，如果持续拉取速度慢于leader副本写入速度，慢于时间超过replica.lag.time.max.ms后，它就变成“非同步”副本，就会被踢出ISR副本集合中。但后面如何follower副本的速度慢慢提上来，那就又可能会重新加入ISR副本集合中了。

而对于ISR副本的观测，是由控制器节点所观测的，当ISR集合发生变化，控制器会通知其他broker节点更新元数据。

消费者组

1. 消息引擎处理模型

   传统的消息引擎处理模型主要有两种，队列模型，和发布-订阅模型。

   • 队列模型：生产者产生消息，就是入队，消费者接收消息就是出队，并删除队列中数据，消息只能被消费一次。但这种模型有一个问题，那就是只能由一个消费者消费，无法直接让多个消费者消费数据。基于这个缺陷，后面又演化出发布-订阅模型。
   • 发布-订阅模型：发布订阅模型中，多了一个主题。消费者会预先订阅主题，生产者写入消息到主题中，只有订阅了该主题的消费者才能获取到消息。这样一来就可以让多个消费者消费数据。

   kafka的消费者组机制，可以同时实现这两种模型。同时还能够对消费组进行动态扩容，让消费变得易于伸缩。

2. 消费者组消费者数量

   所谓消费者组，那自然是由消费者组成的，组内可以有一个或多个消费者实例，而这些消费者实例共享一个id，称为group id。

   消费者组内的所有成员一起订阅某个主题的所有分区，注意一个消费者组中，每一个分区只能由组内的一消费者订阅。

   消费者组内消费者小于或等于分区数，以及topic分区数刚好是消费者组内成员数的倍数。

3. 重平衡

   重平衡其实就是一个协议，它规定了如何让消费者组下的所有消费者来分配topic中的每一个分区。比如一个topic有100个分区，一个消费者组内有20个消费者，在协调者的控制下让组内每一个消费者分配到5个分区，这个分配的过程就是重平衡。

   触发条件：

   • 消费者组内成员发生变更，这个变更包括了增加和减少消费者。注意这里的减少有很大的可能是被动的，就是某个消费者崩溃退出了
   • 主题的分区数发生变更，kafka目前只支持增加分区，当增加的时候就会触发重平衡
   • 订阅的主题发生变化，当消费者组使用正则表达式订阅主题，而恰好又新建了对应的主题，就会触发重平衡

   重平衡过程中，消费者无法从kafka消费消息，这对kafka的TPS影响极大，而如果kafka集内节点较多，比如数百个，那重平衡可能会耗时极多。数分钟到数小时都有可能，而这段时间kafka基本处于不可用状态。所以在实际环境中，应该尽量避免重平衡发生。

总结

本文大致了解了kafka的一些主要结构，实际的执行过程还是要通过生产者和消费者来进行。