Kafka入坑指南

2023-03-15 | 0 评论 | 0 浏览

logo

作为一个Kafka新手，如何入坑？

消息中间件使用场景

1、为什么要使用消息中间件？

在企业级项目开发过程中，经常碰到高并发、微服务交互、异步调用等常见场景，利用消息中间件的削峰、填谷、解耦等特性可以快速实现这些复杂场景，从而实现业务解耦、提升代码的易维护性、提高系统稳定性。

2、哪些常见场景使用消息中间件？

业务场景解耦：比如下单之后发送短信、邮件、站内信等通知......支付之后的异步发货等......数据异步同步等......
大量日志收集：微服务场景下，日志的数据巨大，鼎鼎大名的ELK就是采用Kafka收集日志
高并发场景：比如秒杀场景，大量请求压过来，如果所有流程都放在同一个系统处理，系统基本会宕机。此时上层系统只承担部分任务，然后把请求继续丢到消息队列中让后续系统继续处理，从而快速释放压力继续处理请求。

常用消息中间件

ActiveMQ

Java开发，Apache出品，老牌消息中间件，支持JMS规范，老系统一般都在用。

吞吐量低，只能支持少量消息积压，只有主-备集群模式，可用性稍低。

RocketMQ

阿里出品，Java开发，经历阿里洗礼，经得起考验。

业务系统建议选用RocketMQ，支持消息事务，支持顺序消息，支持消息回溯，吞吐量比Kafka低，分布式集群，可用性高。

RabbitMQ

Erlang开发，笔者接触不多。

Kafka

Apacha出品，Scala开发，处理大数据消息的事实标准，吞吐量极高，分布式集群，可用性高，消息分区，消费者分组。

Kafka入门

Kafka 是目前主流的分布式消息引擎及流处理平台，经常用做企业的消息总线、实时数据管道。早期 Kafka 的定位是一个高吞吐的分布式消息系统，目前则演变成了一个成熟的分布式消息引擎，以及流处理平台。

架构图

Kafka 的设计遵循生产者消费者模式，生产者发送消息到 broker 中某一个 topic 的具体分区里，消费者从一个或多个分区中拉取数据进行消费。

Kafka 依靠 Zookeeper 做分布式协调服务，负责存储和管理 Kafka 集群中的元数据信息，包括集群中的 broker 信息、topic 信息、topic 的分区与副本信息等。

架构图如下：

Kafka架构图

Zookeeper在Kafka中起着关键性作用：

对borker而言
- 状态：zookeeper 记录了所有 broker 的存活状态，broker 会向 zookeeper 发送心跳请求来上报自己的状态。zookeeper 维护了一个正在运行并且属于集群的 broker 列表。
- 控制器选举：kafka 集群中有多个 broker，其中有一个会被选举为控制器。控制器负责管理整个集群所有分区和副本的状态，例如某个分区的 leader 故障了，控制器会选举新的 leader。从多个 broker 中选出控制器，这个工作就是 zookeeper 负责的。
- 限额权限：kafka 允许一些 client 有不同的生产和消费的限额。这些限额配置信息是保存在 zookeeper 里面的。所有 topic 的访问控制信息也是由 zookeeper 维护的。
- 记录 ISR：ISR（in-sync replica）是 partition 的一组同步集合，就是所有 follower 里面同步最积极的那部分。一条消息只有被 ISR 中的成员都接收到，才被视为“已同步”状态。只有处于 ISR 集合中的副本才有资格被选举为 leader。zookeeper 记录着 ISR 的信息，而且是实时更新的，只要发现其中有成员不正常，马上移除。
- node 和 topic 注册：zookeeper 保存了所有 node 和 topic 的注册信息，可以方便的找到每个 broker 持有哪些 topic。node 和 topic 在 zookeeper 中是以临时节点的形式存在的，只要与 zookeeper 的 session 一关闭，他们的信息就没有了。
- topic 配置：zookeeper 保存了 topic 相关配置，例如 topic 列表、每个 topic 的 partition 数量、副本的位置等等。
对consumer而言：
- offset：kafka 老版本中，consumer 的消费偏移量是默认存储在 zookeeper 中的。由于zookeeper的强一致性所以这种方式消费修改延迟高。新版本中，这个工作由 kafka 自己做了，kafka 专门做了一个 offset manager。
- 注册：和 broker 一样，consumer 也需要注册。consumer 会自动注册，注册的方式也是创建一个临时节点，consumer down 了之后就会自动销毁。
- 分区注册：kafka 的每个 partition 只能被消费组中的一个 consumer 消费，kafka 必须知道所有 partition 与 consumer 的关系。

术语

Producer：生产者，消息产生和发送端。
Broker：Kafka 实例，多个 broker 组成一个 Kafka 集群，通常一台机器部署一个 Kafka 实例，一个实例挂了不影响其他实例。
Consumer：消费者，拉取消息进行消费。一个 Topic 可以让若干个消费者（或者消费者实例）进行消费，通常一台机器是一个消费者实例。
Consumer Group：若干个消费者组成一个 Consumer Group 即消费组，一条消息只能被消费组中一个 Consumer 消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量。
Topic：主题，服务端消息的逻辑存储单元。一个 Topic 通常包含若干个 Partition 分区。
Partition：Topic 的分区，分布式存储在各个 broker 中，实现发布与订阅的负载均衡。若干个分区可以被若干个 Consumer 同时消费，达到消费者高吞吐量。一个分区拥有多个副本（Replica），这是Kafka在可靠性和可用性方面的设计。
Replication：每一个分区都有多个副本，副本的作用是做备胎。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。
message：消息，是 Kafka 服务端实际存储的数据，每一条消息都由一个 key、一个 value 以及消息时间戳 timestamp 组成，以文件的形式存储在broker中。
offset：偏移量，分区中的消息位置，由 Kafka 自身维护，Consumer 消费时也要保存一份 offset 以维护消费过的消息位置。

优势 & 特点

Kafka是分布式高扩展架构，吞吐量高，支持数据冗余，支持大量消息积压。

高吞吐、低延时：这是 Kafka 显著的特点，Kafka 能够达到百万级的消息吞吐量，延迟可达毫秒级。
持久化存储：Kafka 的消息最终持久化保存在磁盘之上，提供了顺序读写以保证性能，并且通过 Kafka 的副本机制提高了数据可靠性。
分布式可扩展：Kafka 的数据是分布式存储在不同 broker 节点的，以 topic 组织数据并且按 partition 进行分布式存储，整体的扩展性都非常好。
高容错性：集群中任意一个 broker 节点宕机，Kafka 仍能对外提供服务。

顺序读写、Page Cache、零拷贝技术、分区分段，这些特性让Kafka的读写非常快，也使得Kafka具有了高性能、高吞吐、低延时的特点。

生产-消费方式

Kafka架构图

管理控制台

Kafka-Manager：Kafka的集群管理工具，管理工具可以很容易地发现分布在集群中的哪些topic，以及每个topic的分区情况。

搭建Kafka环境

搭建过程......

Springboot+Kafka示例

项目示例......

消息中间件对比

参考：全面对比 Kafka、RabbitMQ、RocketMQ、ActiveMQ 各自的优缺点

高级特性

Kafka的高级特性

Kafka 消息发送机制

Kafka 生产端发送消息的机制非常重要，这也是 Kafka 高吞吐的基础，生产端的基本流程如下图所示：

640

主要有以下方面的设计：

异步发送
Kafka 自从 0.8.2 版本就引入了新版本 Producer API，新版 Producer 完全是采用异步方式发送消息。生产端构建的 ProducerRecord 先是经过 keySerializer、valueSerializer 序列化后，再是经过 Partition 分区器处理，决定消息落到 topic 具体某个分区中，最后把消息发送到客户端的消息缓冲池 accumulator 中，交由一个叫作 Sender 的线程发送到 broker 端。
这里缓冲池 accumulator 的最大大小由参数 buffer.memory 控制，默认是 32M，当生产消息的速度过快导致 buffer 满了的时候，将阻塞 max.block.ms 时间，超时抛异常，所以 buffer 的大小可以根据实际的业务情况进行适当调整。
批量发送
发送到缓冲 buffer 中消息将会被分为一个一个的 batch，分批次的发送到 broker 端，批次大小由参数 batch.size 控制，默认16KB。这就意味着正常情况下消息会攒够 16KB 时才会批量发送到 broker 端，所以一般减小 batch 大小有利于降低消息延时，增加 batch 大小有利于提升吞吐量。
那么生成端消息是不是必须要达到一个 batch 大小时，才会批量发送到服务端呢？答案是否定的，Kafka 生产端提供了另一个重要参数 linger.ms，该参数控制了 batch 最大的空闲时间，超过该时间的 batch 也会被发送到 broker 端。
消息重试
此外，Kafka 生产端支持重试机制，对于某些原因导致消息发送失败的，比如网络抖动，开启重试后 Producer 会尝试再次发送消息。该功能由参数 retries 控制，参数含义代表重试次数，默认值为 0 表示不重试，建议设置大于 0 比如 3。

Kafka 副本机制

前面提及了 Kafka 分区副本（Replica）的概念，副本机制也称 Replication 机制是 Kafka 实现高可靠、高可用的基础。Kafka 中有 leader 和 follower 两类副本。

Kafka 副本作用

Kafka 默认只会给分区设置一个副本，由 broker 端参数 default.replication.factor 控制，默认值为 1，通常我们会修改该默认值，或者命令行创建 topic 时指定 replication-factor 参数，生产建议设置 3 副本。副本作用主要有两方面：

消息冗余存储，提高 Kafka 数据的可靠性；
提高 Kafka 服务的可用性，follower 副本能够在 leader 副本挂掉或者 broker 宕机的时候参与 leader 选举，继续对外提供读写服务。

关于读写分离

这里要说明的是 Kafka 并不支持读写分区，生产消费端所有的读写请求都是由 leader 副本处理的，follower 副本的主要工作就是从 leader 副本处异步拉取消息，进行消息数据的同步，并不对外提供读写服务。

Kafka 之所以这样设计，主要是为了保证读写一致性，因为副本同步是一个异步的过程，如果当 follower 副本还没完全和 leader 同步时，从 follower 副本读取数据可能会读不到最新的消息。

ISR 副本集合

Kafka 为了维护分区副本的同步，引入 ISR（In-Sync Replicas）副本集合的概念，ISR 是分区中正在与 leader 副本进行同步的 replica 列表，且必定包含 leader 副本。

ISR 列表是持久化在 Zookeeper 中的，任何在 ISR 列表中的副本都有资格参与 leader 选举。

ISR 列表是动态变化的，并不是所有的分区副本都在 ISR 列表中，哪些副本会被包含在 ISR 列表中呢？副本被包含在 ISR 列表中的条件是由参数 replica.lag.time.max.ms 控制的，参数含义是副本同步落后于 leader 的最大时间间隔，默认10s，意思就是说如果某一 follower 副本中的消息比 leader 延时超过10s，就会被从 ISR 中排除。Kafka 之所以这样设计，主要是为了减少消息丢失，只有与 leader 副本进行实时同步的 follower 副本才有资格参与 leader 选举，这里指相对实时。

Kafka 控制器

控制器（Controller）是 Kafka 的核心组件，它的主要作用是在 Zookeeper 的帮助下管理和协调整个 Kafka 集群。集群中任意一个 broker 都能充当控制器的角色，但在运行过程中，只能有一个 broker 成为控制器。

这里先介绍下 Zookeeper，因为控制器的产生依赖于 Zookeeper 的 ZNode 模型和 Watcher 机制。Zookeeper 的数据模型是类似 Unix 操作系统的 ZNode Tree 即 ZNode 树，ZNode 是 Zookeeper 中的数据节点，是 Zookeeper 存储数据的最小单元，每个 ZNode 可以保存数据，也可以挂载子节点，根节点是 /。基本的拓扑图如下：

Zookeeper 有两类 ZNode 节点，分别是持久性节点和临时节点。持久性节点是指客户端与 Zookeeper 断开会话后，该节点依旧存在，直到执行删除操作才会清除节点。临时节点的生命周期是和客户端的会话绑定在一起，客户端与 Zookeeper 断开会话后，临时节点就会被自动删除。

Watcher 机制是 Zookeeper 非常重要的特性，它可以在 ZNode 节点上绑定监听事件，比如可以监听节点数据变更、节点删除、子节点状态变更等事件，通过这个事件机制，可以基于 ZooKeeper 实现分布式锁、集群管理等功能。

控制器选举
当集群中的任意 broker 启动时，都会尝试去 Zookeeper 中创建 /controller 节点，第一个成功创建 /controller 节点的 broker 则会被指定为控制器，其他 broker 则会监听该节点的变化。当运行中的控制器突然宕机或意外终止时，其他 broker 能够快速地感知到，然后再次尝试创建 /controller 节点，创建成功的 broker 会成为新的控制器。
控制器功能
前面我们也说了，控制器主要作用是管理和协调 Kafka 集群，那么 Kafka 控制器都做了哪些事情呢，具体如下：
- 主题管理：创建、删除 topic，以及增加 topic 分区等操作都是由控制器执行。
- 分区重分配：执行 Kafka 的 reassign 脚本对 topic 分区重分配的操作，也是由控制器实现。
- Preferred leader 选举：这里有一个概念叫 Preferred replica 即优先副本，表示的是分配副本中的第一个副本。Preferred leader 选举就是指 Kafka 在某些情况下出现 leader 负载不均衡时，会选择 preferred 副本作为新 leader 的一种方案。这也是控制器的职责范围。
- 集群成员管理：控制器能够监控新 broker 的增加，broker 的主动关闭与被动宕机，进而做其他工作。这里也是利用前面所说的 Zookeeper 的 ZNode 模型和 Watcher 机制，控制器会监听 Zookeeper 中 /brokers/ids 下临时节点的变化。
- 数据服务：控制器上保存了最全的集群元数据信息，其他所有 broker 会定期接收控制器发来的元数据更新请求，从而更新其内存中的缓存数据。
从上面内容我们大概知道，控制器可以说是 Kafka 的心脏，管理和协调着整个 Kafka 集群，因此控制器自身的性能和稳定性就变得至关重要。
社区在这方面做了大量工作，特别是在 0.11 版本中对控制器进行了重构，其中最大的改进把控制器内部多线程的设计改成了单线程加事件队列的方案，消除了多线程的资源消耗和线程安全问题，另外一个改进是把之前同步操作 Zookeeper 改为了异步操作，消除了 Zookeeper 端的性能瓶颈，大大提升了控制器的稳定性。

Kafka消费端Rebalance机制

前面介绍消费者术语时，提到了消费组的概念，一个 topic 可以让若干个消费者进行消费，若干个消费者组成一个 Consumer Group 即消费组，一条消息只能被消费组中的一个消费者进行消费。我们用下图表示Kafka的消费模型。

Rebalance 概念

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

Rebalance 发生条件

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

消费组的消费者成员数量发生变化
消费主题的数量发生变化
消费主题的分区数量发生变化

其中后两种情况一般是计划内的，比如为了提高消息吞吐量增加 topic 分区数，这些情况一般是不可避免的，后面我们会重点讨论如何避免因为组内消费者成员数发生变化导致的 Rebalance。

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 实例，负责消费组下各个消费者和服务端组协调器之前的通信。可以用下图表示协调器原理：