Kafka消息队列入门到精通看这一篇就够了

文章目录

第一章概述
- 1.1 Kafka 的定义及特点
- 1.2 消息队列的介绍
- 1.3 Kafka 的基础架构
第二章入门
- 2.1 Kafka 的安装部署
- 2.2 Kafka 命令行操作
第三章架构深入
- 3.1 Kafka 工作流程
- 3.2 Kafka 文件存储机制
- 3.3 Kafka 生产者
- 3.4 Kafka 消费者
- 3.5 Kafka 高效读写数据
- 3.6 Kafka 事务
第四章 API
第五章监控
第六章 Flume 对接 Kafka
第七章 Kafka 面试题

第一章概述

1.1 Kafka 的定义及特点

一个分布式的，基于发布/订阅模式的消息队列（Massage Queue），主要应用于大数据实时处理领域；
Kafka 中数据是有时效性地保存在磁盘中；
Kafka 由 Scala 编写；

1.2 消息队列的介绍

有两种处理系统任务的方式：同步处理和异步处理。
同步处理：每一步必须等到前一步完成，才可执行。
异步处理：分批次处理，前一步未完成也可以开始下一步。
消息队列为异步处理，使用消息队列的好处：
- 解耦：只要遵循相同接口，则可以独立扩展或修改不同处理过程，提高可恢复性和健壮性。
- 缓冲：有助于控制和优化数据经过系统的速度，解决生产者和消费者速度不一致的问题。衍生灵活性和削峰能力，以低成本应对突变的任务量。
- 异步通信：用户需求，有时不需要立即处理，而是存储在队列中，需要时使用，如邮箱。
消息队列的两种模式：
- 点对点：
  》消费者主动拉取数据，收到后队 Queue 即删除该数据，无法消费曾消费过的数据；
  》Queue 支持多个消费者，但是对于一个消息，只能有一个消费者；
  》消费者主动拉取数据，生产者不好控制推送速度。
- 发布订阅：
  
  》消费者消费后的数据不会从 Queue 中删除；
  》生产者生产消息到 topic 中，所有订阅该 topic 的消费者都会收到该消息；
  》队列主动推送/消费者主动拉取，两种传递模式。需要长期维持轮询。

1.3 Kafka 的基础架构

生产：
1. ProducerA 生产消息，此消息属于TopicA，被分为若干分区，发送至 KafkaCluster 中的一个节点 Broker1 的TopicA.Partition0.Leader，消息的另一部分被发往 Broker2 节点的 TopicA.Partition1。
2. 一个 broker 可以容纳多个 topic；逻辑层的 Topic。
3. 一个 topic 可分为多个 partition，partition 是一个有序队列，实现扩展性；物理层的 Partition。
4. 每个分区有对于的 Leader 和 Follower
  Leader 是 producer 发送数据的对象、Comsumer 消费数据的对象，一个 Leader 对应多个 Follower；
  Follower 用于存储所属分区的副本，实时从 Leader 中同步数据，Leader 故障时多个 Follower 选一个成为新的 Leader。
消费：
1. 一个消费者组 CG 逻辑上是一个订阅者，消费一个 Topic，消费者组间互不影响。
2. 一个分区只能由一个组中的一个消费者消费，一个消费者可以消费该组所属 Topic 的多个分区，因此，当一个组中的消费者多余该组对应 Topic 的 Partition 数时，无意义；而当两者相等时，消费率最高。
3. 每个消费者都属于一个 CG。
4. 消费者组中的成员，分别从 KafkaCluster 的不同节点拉取 Topic 的不同分区。
5. 一个 CG 中的一个消费者，可以订阅不同的 Topic。
offset 与 Zookeeper
1. Kafka 集群和消费者都会向 ZK 注册。0.9 版本前 offset 存在 ZK，0.9 及之后 offset 存在 KafkaCluster，因为消费者消费速度非常快，若加之于 ZK 的连接通信，这样的高并发对 ZK 和 Kafka都不好。offset 用于存储当前消费的数据在整体中的偏移量；
2. offset 主题分为 50 个分区，副本数只有一份；
3. Kafka 集群的 brokers 和 comsumers 都会向 Zookeeper 注册自己。
Replica
1. 为保证数据的可靠性，每个分区都有若干 Replica；
2. Kafka 中的副本数不能超过当前的可用的 broker 数，多余的副本会放在同一个节点上；
3. 而 HDFS 中如果副本数大于 workers 数量，实际副本数为 workers 数量。
分区
1. MapReduce 中的分区是为提高 ReduceTask 的并发度，提高计算效率；
2. Hive 中的分区是为了查询时，减少读取的数据量，提高查询效率；
3. Kafka 中的分区是为了提高某个 Topic 的负载能力，提高集群的负载均衡，提高数据传输并发度，和可扩展性（提高 brokers 的水平扩展，也可看作并发度的扩展）
4. 有了分区，为何还要 Segment ？
  若没 Segment，一个 Partition 对应一个文件，文件会持续增大。Data Purge 定期执行时，需要把当前的和旧数据一并删除，然后创建一个块新的文件（物理位置）；这不符合 Kafka 对数据在物理层顺序写入的优化初衷。而引入 Segment 后，每次 Data Purge 只需把旧的 Segment 删除，保证当前数据在物理层始终是顺序写入，提高效率。

第二章入门

2.1 Kafka 的安装部署

下载
1. 官方下载地址
  
  http://kafka.apache.org/downloads.html
2. 版本选择
  
  kafka_2.11-2.4.1.tgz
集群规划
---- hadoop102 ----|---- hadoop103 ----|---- hadoop104 ----
-------- zk ---------|--------- zk --------|-------- zk --------
------ kafka --------|------- kafka -------|------ kafka -------

安装部署

# 解压安装包
tar -zxvf /opt/software/kafka_2.11-2.4.1.tgz -C /opt/module/# 修改安装目录
mv kafka_2.11-2.4.1 kafka# 创建日志文件夹，注意这里也会存储 topic 数据
cd $KAFKA_HOME; mkdir logs# 修改配置文件，配置如下
vim $KAFKA_HOME/config/server.properties

# broker 的全局唯一编号，不能重复
broker.id=102# 删除 topic 功能使能，当前版本默认 true，且隐藏此项
delete.topic.enable=true# 处理网络请求的线程数量
num.network.threads=3# 用来处理磁盘 IO 的线程数量
num.io.threads=8# 发送套接字的缓冲区大小
socket.send.buffer.bytes=102400# 接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400# 请求套接字的缓冲区大小 100M
socket.request.max.bytes=104857600# Kafka 运行日志存放路径
log.dirs=/opt/module/kafka/logs# topic 在当前 broker 上的分区个数
num.partitions=1# 用来恢复和清理 data 下数据的线程数量
num.recovery.threads.per.data.dir=1# segment 文件保留的最长时间，超时将被删除 1week
log.retention.hours=168# 必须配置项 Zookeeper 集群地址
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka

# 配置环境变量
sudo vim /etc/profile.d/my_env.sh# KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:KAFKA_HOME/bin# 立即生效环境变量
source /etc/profile# 其他服务器
- 分发 /opt/module/Kafka
- 配置环境变量
- 修改 server.properties 中 broker.id

启动和关闭

先启动 Zookeeper，然后启动 Kafka

# 各个节点分别启动：kafka-server-start.sh -daemon config/server.properties-daemon 后台启动# 各个节点分别关闭：kafka-server-stop.sh stop

启动脚本

#!/bin/bash
case $1 in
"start")for i in hadoop102 hadoop103 hadoop104doecho "============== $i Kafka =============="ssh $i /opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.propertiesdone
;;
"stop")for i in hadoop102 hadoop103 hadoop104doecho "============== $i Kafka =============="ssh $i /opt/module/kafka/bin/kafka-server-stop.shdone
;;
*)echo "Input Args Error..."
;;
esac

关闭：
注意：关闭 Kafka 集群的操作有一定的持续时间，如果在此时关闭了 Zookeeper，此时没有关闭成功的 Kafka 服务只能用 kill -9 来关闭了。

2.2 Kafka 命令行操作

查看所有 Topic

# 其中 hadoop102:9092 表示服务入口，可写其他节点，可写多个保证连接可靠性；
# --bootstrap-server 表示数据偏移量存储在 Kafka 中；如果参数是 --zookeeper 则数据偏移量存储在 zookeeper 中；0.9 及以上版本 Kafka 为 前者；
kafka-topics.sh --bootstrap-server hadoop102:9092 --list

创建 Topic

kafka-topics.sh  --bootstrap-server hadoop102:9092 --create --replication-factor 2 --partitions 1 --topic 主题名--replication-factor 副本数
--partitions 分区数
--topic 主题名

删除 Topic

kafka-topics.sh --bootstrap-server hadoop102:9092 --delete --topic 主题名

生产消息

# --broker-list 指定接收数据的服务端；kafka-console-producer.sh --broker-list hadoop102:9092 --topic 主题名
>hello world
>hello kafka

消费消息

# 消费从此刻开始的新数据
kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic 主题名# 消费指定主题的所有数据
kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic 主题名

查看指定 Topic 详情

kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic 主题名

修改分区数

kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic 主题名 --partitions 修改后分区数

第三章架构深入

3.1 Kafka 工作流程

 1. 生产者向 Kafka 集群中已有的 Topic 发送数据；2. Topic 中信息被分成多干分区，每一条接收到的数据都存在 Partition 下的文件中，并不断追加到对应文件末端，每条数据都有自己的 offset3. 每个分区的 Leader 将数据副本发送至 Follower；4. 消费者从指定主题的分区中消费数据，并记录自己消费到的数据的 offset，以便出错恢复时，从上一次的位置继续消费。

3.2 Kafka 文件存储机制

由于生产者生产的消息会不断追加到 log 文件末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采用分片和索引机制，将每个 partition 分为多个 segment。
每个 segment 对应两个文件—— .index 和 .log 文件。
一个分区对应一个文件夹。文件夹命名规则为：主题名+分区号。例如，first 这个主题有三个分区，其对应的文件夹为：first-0，first-1，first-2；本文配置的数据存储在 /opt/module/kafka/logs
```
00000000000000000000.index
00000000000000000000.log
00000000000000170410.index
00000000000000170410.log
00000000000000239430.index
00000000000000239430.log
```
index 和 log 文件以当前的 segmant 中的第一条消息的 offset 命名

3.3 Kafka 生产者

分区策略

为什么要分区？
》方便数据在集群中的扩展：每个 Partition 可以调节以适应所在的机器，整个集群可以适应任意大小的数据；

》可以提高数据读写的并发度：以 Partition 的 Leader 为读写单位。

分区规则

// producer 推送的数据被封装成一个 ProducerRecord 对象// ProducerRecord 类的构造器
ProducerRecord(@NotNull String topic, Integer partition, Long timestamp, String key, String value, @Nullable Iterable<Header> headers)/*
1. 指明 parition 的情况下，直接将指明的值作为 partition 的值；
2. 没指明 partition 但有 key，将 key 的 hash 值与 topic 的分区数进行求余运算得 parition 值；
3. partition 与 key 都没有指定，Kafka 采用 StickyPartition(粘性分区器)，随机选择一个分区，并尽可能的一直使用该分区，待该分区的 batch 数据批已满或者超过时间间隔，Kafka 再随机一个分区进行写入。
*/

数据可靠性保证

生产者 pull 数据到 Leader 的可靠性保证

》每个 partition 收到 producer 的数据后，需要回复 ack；
》producer 收到当前消息的 ack 之后，才会继续发送，否则重复发送；

》确保有 Follower 与 Leader 同步完成后，Leader 才回复 ack；这样能确保 Leader 挂掉 Follower 替代 Leader 时没有数据丢失；

》多少个 Follower 同步完成 Leader 回复 ack？方案1:半数以上；方案2:all

Topic Partition 保存数据的可靠性

》Follower 副本同步

方案	特点
半数以上	延迟低
all	延迟高

m 个 broker 的集群，需要 n 台节点的容错能力，假设副本数为 x
—方案1：最坏的情况，n 台节点挂掉，并丢失 n 个副本，此时若要返回 ack 则需要有 x/2<n+1 个副本已经同步成功，即配置 x>= 2n+1 时，才能保证等于或低于 n 台节点挂掉，仍能使 ack 顺利返回，数据继续发送；这种方案数据大量冗余；
—方案2：最坏的情况，n 台节点挂掉，并丢失 n 个副本，此时若要返回 ack 则需要 x 个副本全部同步成功，即配置任意副本数，都无法使 ack 顺利返回，数据无法继续发送；

— Kafka 既没有选择方案1，也没有选择方案2；

》ISR

》ack 应答级别
对于可靠性要求不高的数据，比如前端埋点的数据能够容忍数据的少量丢失，没必要等 ISR 中的 Follower 全部接受成功。所以 Kafka 提供了三种可靠性级别，用户根据对数据可靠性和延迟的要求，做出相应的配置。

Exactly Once

3.4 Kafka 消费者

3.5 Kafka 高效读写数据

3.6 Kafka 事务

第四章 API

第五章监控

第六章 Flume 对接 Kafka

第七章 Kafka 面试题