一文了解 Kafka 分布式！

点击上方 "程序员小乐"关注, 星标或置顶一起成长

每天凌晨00点00分, 第一时间与你相约

每日英文

Sometimes, it is better to be alone. That way, nobody can hurt you.

有时候，孤身一人反而更好。那样，就没有人能伤害你了。

每日掏心话

人生匆匆，若不时而驻足欣赏，将会错过它的美好。感情里，最缺的不是激情，而是激情过后的那份坚持。

来自：bainianminguo | 责编：乐乐

链接：cnblogs.com/bainianminguo/p/12247158.html

程序员小乐(ID:study_tech)第 790 次推文   图片来自百度

往日回顾：又一个程序员“删库跑路”被抓了，一己之力蒸发公司市值超10亿，300万商铺遭瘫痪

   正文   

1. Kafka 概述

1.1 定义

Kakfa 是一个分布式的基于发布/订阅模式的消息队列（message queue），主要应用于大数据的实时处理领域。

1.2 消息队列

1.2.1 传统消息队列与新式消息队列模式

上面是传统的消息队列，比如一个用户要注册信息，当用户信息写入数据库后，后面还有一些其他流程，比如发送短信，则需要等这些流程处理完成后，再返回给用户。而新式队列，比如一个用户注册信息，数据直接丢进数据库，就直接返回给用户成功。

1.2.2 使用消息队列的好处

• 解耦

• 可恢复性

• 缓冲

• 灵活性与峰值处理能力

• 异步通信

1.2.3 消息队列的模式

1) 点对点模式

消息生产者发送消息到消息队列中，然后消息消费者从队列中取出并且消费消息，消息被消费后，队列中不在存储。所以消息消费者不可能消费到已经被消费的消息；队列支持存在多个消费者，但是对于一个消息而言，只会 有一个消费者可以消费；如果想发给多个消费者，则需要多次发送该条消息。

2) 发布/订阅模式（一对多，消费者消费数据之后不会清除消息）

消息生产者将消息发布到 topic 中，同时有多个消息消费者（订阅）消费该消息。和点对点的方式不同，发布到 topic 的消息会被所有的订阅者消费；但是数据保留是期限的，默认是 7 天，因为它不是存储系统。Kafka 就是这种模式的。有两种方式，一种是是消费者去主动去消费（拉取）消息，而不是生产者推送消息给消费者；另外一种就是生产者主动推送消息给消费者，类似公众号。

1.3 Kafka 基础架构

Kafka 的基础架构主要有 broker、生产者、消费者组构成，当前还包括 ZooKeeper。

生产者负责发送消息，broker 负责缓冲消息，broker 中可以创建 topic，每个 topic 又有 partition 和 replication 的概念。

消费者组负责处理消息，同一个消费者组的中消费者不能消费同一个 partition 中的数据。消费者组主要是提高消费能力，比如之前是一个消费者消费 100 条数据，现在是 2 个消费者消费 100 条数据，可以提高消费能力。所以消费者组的消费者的个数要小于 partition 的个数，不然就会有消费者没有 partition 可以消费，造成资源的浪费。

注意：不同消费者组的消费者是可以消费相同的 partition 数据。

Kakfa 如果要组件集群，则只需要注册到一个 ZooKeeper 中就可以了，ZooKeeper 中还保留消息消费的进度或者说偏移量或者消费位置。

• 0.9 之前的版本偏移量存储在 ZooKeeper；

• 0.9 之后的版本偏移量存储在 Kafka中。Kafka 定义了一个系统 topic，专用用来存储偏移量的数据。

为什么要改？

主要是考虑到频繁更改偏移量，对 ZooKeeper 的压力较大，而且 Kafka 本身自己的处理也较复杂。

1.4 安装 Kafka

1) Kafka 的安装只需要解压安装包就可以完成安装。

tar -zxvf kafka_2.11-2.1.1.tgz -C /usr/local/

2) 查看配置文件。

[root@es1 config]# pwd
/usr/local/kafka/config
[root@es1 config]# ll
total 84
-rw-r--r--. 1 root root  906 Feb  8  2019 connect-console-sink.properties
-rw-r--r--. 1 root root  909 Feb  8  2019 connect-console-source.properties
-rw-r--r--. 1 root root 5321 Feb  8  2019 connect-distributed.properties
-rw-r--r--. 1 root root  883 Feb  8  2019 connect-file-sink.properties
-rw-r--r--. 1 root root  881 Feb  8  2019 connect-file-source.properties
-rw-r--r--. 1 root root 1111 Feb  8  2019 connect-log4j.properties
-rw-r--r--. 1 root root 2262 Feb  8  2019 connect-standalone.properties
-rw-r--r--. 1 root root 1221 Feb  8  2019 consumer.properties
-rw-r--r--. 1 root root 4727 Feb  8  2019 log4j.properties
-rw-r--r--. 1 root root 1925 Feb  8  2019 producer.properties
-rw-r--r--. 1 root root 6865 Jan 16 22:00 server-1.properties
-rw-r--r--. 1 root root 6865 Jan 16 22:00 server-2.properties
-rw-r--r--. 1 root root 6873 Jan 16 03:57 server.properties
-rw-r--r--. 1 root root 1032 Feb  8  2019 tools-log4j.properties
-rw-r--r--. 1 root root 1169 Feb  8  2019 trogdor.conf
-rw-r--r--. 1 root root 1023 Feb  8  2019 zookeeper.properties

3) 修改配置文件 server.properties。

设置 broker.id 这个是 Kafka 集群区分每个节点的唯一标志符。

4) 设置 Kafka 的数据存储路径。

注意：这个目录下不能有其他非 Kafka 目录，不然会导致 Kafka 集群无法启动。

5) 设置是否可以删除 topic，默认 Kafka 的 topic 是不允许删除的。

6) Kafka 的数据保留的时间，默认是 7 天。

7) Log 文件最大的大小，如果 log 文件超过 1 G 会创建一个新的文件。

8) Kafka 连接的 ZooKeeper 的地址和连接 Kafka 的超时时间。

9) 默认的 partition 的个数。

1.5 启动 Kafka

1) 启动方式一，Kafka 只能单节点启动，所以每个 Kakfa 节点都需要手动启动，下面的方式阻塞的方式启动。

2) 启动方式二，守护的方式启动，推荐使用。

1.6 Kafka 操作

1) 查看当前 Kafka 集群已有的 topic。

注意：这里连接的 ZooKeeper，而不是连接的 Kafka。

2) 创建 topic，指定分片和副本个数。

说明：replication-factor 副本数，replication-factor 分区数，topic 主题名。

如果当前 Kafka 集群只有 3 个 broker 节点，则 replication-factor 最大就是 3 了，下面的例子创建副本为 4，则会报错。

3) 删除 topic。

4） 查看 topic 信息。

1.7 启动生产者生产消息，Kafka 自带一个生产者和消费者的客户端

1) 启动一个生产者，注意此时连的 9092 端口，连接的 Kafka 集群。

2) 启动一个消费者，注意此时连接的还是 9092 端口，在 0.9 版本之前连接的还是 2181 端口。

这里我们启动 2 个消费者来测试一下。

说明：如果不指定的消费者组的配置文件的话，默认每个消费者都属于不同的消费者组。

3) 发送消息，可以看到每个消费者都能收到消息。

4）Kakfa 中的实际数据。

2. Kafka 架构深入

Kafka 不能保证消息的全局有序，只能保证消息在 partition 内有序，因为消费者消费消息是在不同的 partition 中随机的。

2.1 Kafka 的工作流程

Kafka 中的消息是以 topic 进行分类的，生产者生成消息、消费者消费消息都面向 topic。

Topic 是一个逻辑上的概念，而 partition 是物理上的概念。每个 partition 又有副本的概念。每个 partition 对应于一个 log 文件，该 log 文件中存储的就是生产者生成的数据，生产者生成的数据会不断的追加到该 log 的文件末端，且每条数据都有自己的 offset，消费者都会实时记录自己消费到了那个 offset，以便出错的时候从上次的位置继续消费，这个 offset 就保存在 index 文件中。Kafka 的 offset 是分区内有序的，但是在不同分区中是无顺序的，Kafka 不保证数据的全局有序。

2.2 Kafka 原理

由于生产者生产的消息会不断追加到 log 文件的末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采用分片和索引的机制，将每个 partition 分为多个 segment，每个 segment 对应2个文件 — index 文件和 log 文件，这 2 个文件位于一个相同的文件夹下，文件夹的命名规则为：topic 名称 + 分区序号。

Index 和 log 的文件的文件名是当前这个索引是最小的数据的 offset。Kafka 如何快速的消费数据呢？

Index 文件中存储的数据的索引信息，第一列是 offset，第二列这这个数据所对应的 log 文件中的偏移量，就像我们去读文件，使用 seek() 设置当前鼠标的位置一样，可以更快的找到数据。

如果要去消费 offset 为 3 的数据，首先通过二分法找到数据在哪个 index 文件中，然后在通过 index 中 offset 找到数据在 log 文件中的 offset；这样就可以快速的定位到数据，并消费。

所以，Kakfa 虽然把数据存储在磁盘中，但是他的读取速度还是非常快的。

3. Kafka 生产者和消费者

3.1 Kafka 生产者

Kafka 的 partition 分区的作用

Kafka 分区的原因主要就是提供并发提高性能，因为读写是 partition 为单位读写的；那生产者发送消息是发送到哪个 partition 中呢？

1. 在客户端中指定 partition；

2. 轮询（推荐）消息1去 p1，消息2去 p2，消息3去 p3，消息4去 p1，消息5去 p2，消息6去 p3……

3.2 Kafka 如何保证数据可靠性呢？通过 ack 来保证

为保证生产者发送的数据，能可靠的发送到指定的 topic，topic 的每个 partition 收到生产者发送的数据后，都需要向生产者发送 ack（确认收到），如果生产者收到 ack，就会进行下一轮的发送，否则重新发送数据。

那么 Kafka 什么时候向生产者发送 ack？

确保 follower 和 leader 同步完成，leader 在发送 ack 给生产者，这样才能确保 leader 挂掉之后，能在 follower 中选举出新的 leader 后，数据不会丢失。

那多少个 follower 同步完成后发送 ack？

• 方案1：半数已经完成同步，就发送 ack；

• 方案2：全部完成同步，才发送 ack（Kafka采用这种方式）

采用第二种方案后，设想以下场景：leader 收到数据，所有的 follower 都开始同步数据，但是有一个 follower 因为某种故障，一直无法完成同步，那 leader 就要一直等下，直到他同步完成，才能发送 ack。这样就非常影响效率，这个问题怎么解决？

Leader 维护了一个动态的 ISR 列表（同步副本的作用），只需要这个列表的中的 follower 和 leader 同步；当 ISR 中的 follower 完成数据的同步之后，leader 就会给生产者发送 ack，如果 follower 长时间未向 leader 同步数据，则该 follower 将被剔除ISR，这个时间阈值也是自定义的；同样 leader 故障后，就会从 ISR 中选举新的 leader。

怎么选择 ISR 的节点呢？

首先通信的时间要快，要和 leader 要可以很快的完成通信，这个时间默认是 10s

然后就看 leader 数据差距，消息条数默认是 10000 条（后面版本被移除）

为什么移除：因为 Kafka 发送消息是批量发送的，所以会一瞬间 leader 接受完成，但是 follower 还没有拉取，所以会频繁踢出和加入ISR，这个数据会保存到 ZooKeeper 和内存中，所以会频繁更新 ZooKeeper 和内存。

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

1) acks 为 0

生产者不等 ack，只管往 topic 丢数据就可以了，这个丢数据的概率非常高。

2) ack为 1

leader 落盘后就会返回 ack，会有数据丢失的现象，如果 leader 在同步完成后出现故障，则会出现数据丢失。

3) ack为-1（all）

leader 和 follower（ISR）落盘才会返回 ack，会有数据重复现象，如果在 leader 已经写完成，且 follower 同步完成，但是在返回ack的出现故障，则会出现数据重复现象；极限情况下，这个也会有数据丢失的情况，比如 follower 和 leader 通信都很慢，所以 ISR 中只有一个 leader 节点，这个时候，leader 完成落盘，就会返回 ack，如果此时 leader 故障后，就会导致丢失数据。

3.3 Kafka 如何保证消费数据的一致性？通过HW来保证

• LEO：指每个 follower 的最大的 offset；

• HW（高水位）：指消费者能见到的最大的 offset，LSR 队列中最小的 LEO，也就是说消费者只能看到1~6的数据，后面的数据看不到，也消费不了。

避免 leader 挂掉后，比如当前消费者消费8这条数据后，leader 挂了，此时比如 f2 成为 leader，f2 根本就没有9这条数据，那么消费者就会报错，所以设计了 HW 这个参数，只暴露最少的数据给消费者，避免上面的问题。

3.3.1 HW 保证数据存储的一致性

1) follower 故障

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

2) leader 故障

leader 发生故障后，会从 ISR 中选出一个新的 leader，之后，为了保证多个副本之间的数据一致性，其余的 follower 会先将各自的 log 文件高于 hw 的部分截掉（新 leader 自己不会截掉），然后从新的 leader 同步数据。

注意：这个是为了保证多个副本间的数据存储的一致性，并不能保证数据不丢失或者不重复。

3.3.2 精准一次（幂等性），保证数据不重复

• ack 设置为 -1，则可以保证数据不丢失，但是会出现数据重复（at least once）

• ack 设置为 0，则可以保证数据不重复，但是不能保证数据不丢失（at most once）

但是如果鱼和熊掌兼得，该怎么办？这个时候就就引入了 Exact once（精准一次）。

在 0.11 版本后，引入幂等性解决 Kakfa 集群内部的数据重复，在 0.11 版本之前，在消费者处自己做处理。如果启用了幂等性，则 ack 默认就是-1，Kafka 就会为每个生产者分配一个 pid，并未每条消息分配 seqnumber，如果 pid、partition、seqnumber 三者一样，则 Kafka 认为是重复数据，就不会落盘保存；但是如果生产者挂掉后，也会出现有数据重复的现象；所以幂等性解决在单次会话的单个分区的数据重复，但是在分区间或者跨会话的是数据重复的是无法解决的。

3.4 Kafka 消费者

3.4.1 消费方式

消息队列有两种消费消息的方式，push（微信公众号）、pull（kafka）。push 模式很难适应消费速率不同的消费者，因为消费发送速率是由 broker 决定的，他的目标是尽可能以最快的的速度传递消息，但是这样很容易造成消费者来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而 pull 的方式可以消费者的消费能力以适当的速率消费消息。

pull 模式的不足之处是如果 Kafka 没有数据，消费者可能会陷入死循环，一直返回空数据，针对这一点，Kafka 消费者在消费数据时候回传递一个 timeout 参数，如果当时没有数据可供消费，消费者会等待一段时间在返回。

3.4.2 分区分配策略

一个消费者组有多个消费者，一个 topic 有多个 partition。所以必然会涉及到 partition 的分配问题，即确定哪个 partition 由哪个消费者来消费。Kafka 提供两种方式，一种是轮询（RountRobin）对于 topic 组生效，一种是（Range）对于单个topic生效

轮询：前置条件是需要一个消费者里的消费者订阅的是相同的 topic。不然就会出现问题；非默认的的方式。

同一个消费者组里的消费者不能同时消费同一个分区，比如三个消费者消费一个 topic 的 9 个分区。

如果一个消费者组里有2个消费者，这个消费者组里同时消费 2 个 topic，每个 topic 又有三个 partition。首先会把 2 个 topic 当做一个主题，然后根据 topic 和 partition 做 hash，然后在按照 hash 排序。然后轮询分配给一个消费者组中的 2 个消费者。

如果是下面这样的方式订阅的呢？

比如有 3 个 topic，每个 topic 有 3 个 partition，一个消费者组中有 2 个消费者。消费者1订阅 topic1 和 topic2，消费者2订阅 topic2 和 topic3。那么这样的场景，使用轮训的方式订阅 topic 就会有问题。

如果是下面这种方式订阅呢？

比如有2个 topic，每个 topic 有3个 partition，一个消费者组 有2个消费者，消费者1订阅 topic1，消费者2订阅 topic2，这样使用轮训的方式订阅 topic 也会有问题。

所以我们一直强调，使用轮训的方式订阅 topic 的前提是一个消费者组中的所有消费者订阅的主题是一样的；所以轮询的方式不是 Kafka 默认的方式；Range 是按照单个 topic 来划分的，默认的分配方式。

Range 的问题会出现消费者数据不均衡的问题。比如下面的例子，一个消费者组订阅了 2 个 topic，就会出现消费者1消费 4 个 partition，而另外一个消费者只消费 2 个 partition。

分区策略什么时候会触发呢？当消费者组里的消费者个数变化的时候，会触发分区策略调整，比如消费者里增加消费者，或者减少消费者。

3.4.3 维护 offset

由于消费者在消费过程中可能会出现断电宕机等故障，消费者恢复后，需要从故障前的位置继续消费，所以消费者需要实施记录自己消费哪个 offset，以便故障恢复后继续消费。

Offset保存的位置有2个，一个 ZooKeeper，一个是 Kafka。首先看下 offset 保存到 ZooKeeper，由消费者组、topic、partition 三个元素确定唯一的 offset。

所以消费者组中的某个消费者挂掉之后，或者的消费者还是可以拿到这个 offset。

Controller 这个节点和 ZooKeeper 通信，同步数据，这个节点就是谁先起来，谁就先注册 controller，谁就是 controller。其他节点和 controller 信息保持同步。

3.4.5 消费者组的案例

修改消费者组 id

启动一个消费者发送 3 条数据。

指定消费者组启动消费者，启动三个消费者，可以看到每个消费者消费了一条数据。

在演示下不同组可以消费同一个 topic 的，我们看到 2 个消费者的消费者都消费到同一条数据。再次启动一个消费者，这个消费者属于另外一个消费者组。

4 Kafka 的高效读写机制

4.1 分布式部署

多节点并行操作。

4.2、顺序写磁盘

Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程中一直追加到文件末尾，为顺序写，官网有数据表明。同样的磁盘，顺序写能到 600M/S，而随机写只有 100K/S。这与磁盘的机械结构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。

4.3、零复制技术

正常情况下，先把数据读到内核空间，在从内核空间把数据读到用户空间，然后在调操作系统的 IO 接口写到内核空间，最终在写到硬盘中。

Kafka 是这样做的，直接在内核空间流转 IO 流，所以 Kafka 的性能非常高。

5. ZooKeeper 在 Kafka 中的作用

Kafka 集群中有一个 broker 会被选举为 controller，负责管理集群 broker 的上下线，所有的 topic 的分区副本分配和 leader 选举等工作。

欢迎在留言区留下你的观点，一起讨论提高。如果今天的文章让你有新的启发，学习能力的提升上有新的认识，欢迎转发分享给更多人。

欢迎各位读者加入订阅号程序员小乐技术群，在后台回复“加群”或者“学习”即可。

猜你还想看

关注订阅号「程序员小乐」，收看更多精彩内容

嘿，你在看吗？