从零开始的 Kafka 学习(三)| 创建主题

发布于:2025-03-17 ⋅ 阅读:(19) ⋅ 点赞:(0)

1. 创建主题

Topic 主题是 Kafka 中消息的逻辑分类,但是这个分类不应该是固定的,而是应该由外部的业务场景进行定义(注意:Kafka 中其实是有两个固定的,用于记录消费者偏移量和事务处理的主题),所以 Kafka 提供了相应的指令和客户端进行主题操作。

1.1 相关概念

1.1.1 主题:Topic

Kafka 是分布式消息传输系统,采用的数据传输方式为发布/订阅模式,也就是说由消息的生产者发布消息,消费者订阅消息后获取数据。为了对消费者订阅的消息进行区分,所以对消息在逻辑上进行了分类,这个分类我们称之为主题:Topic。消息的生产者必须将消息数据发送到某一个主题,而消费者必须从某一个主题中获取消息,并且消费者可以同步消息一个或多个主题的数据。Kafka 集群中可以存放主题的消息数据。

1.1.2 分区:Partition

Kafka 消息传输采用发布/订阅模式,所以消息生产者必须将数据发送到一个主题,家人u发送给这个主题的数据非常多,那么主题所在 broker 节点的负载和吞吐量就会受到极大的考验,甚至有可能因为热点问题引起 broker 节点故障,导致服务不可用。一个好的方案就是将一个主题从物理上分成几块,然后将不同的护具块均匀地分配到不同的 broker 节点上,这样就可以缓解单节点的负载问题。这个主题的分块我们称之为:分区 partition。默认情况下,topic 主题创建时分区数量为1,也就是一块分区,可以指定参数–partitions改变。Kafka 的分区解决了单一主题 topic 线性扩展的问题,也解决了负载均衡问题。

topic 主题的每个分区都会用一个编号进行标记,一般是从0开始的连续整数数字,Partition 分区是物理上的概念,也就意味着以数据文件的方式真实存在。每个 topic 包含一个或多个 parition,每一个 partition 都是一个有序的队列。partition 中每条消息都会分配一个有序的ID,称之为偏移量:Offset。

在这里插入图片描述

1.1.3 副本:Replication

分布式系统出现错误是比较常见的,只要保证集群内部依然存在可用的服务节点即可,当然效率会有所降低,不过只要能保证系统可用就可以了。Kafka 的topic也存在类似的问题,也就是说,如果一个topic 划分了多个分区 partition,那么这些分区怒就会均匀地分布在不同的 broker 节点上,一旦某一个 broker 节点出现了问题,那么在这个节点上的分区就会出现问题,那么 Topic 的数据就不完整了。所以一般情况下,为了防止出现数据丢失的情况,我们会给分区数据设定多个备份。这里的备份,我们称之为:副本 Replication。

Kafka 支持多副本,使得主题 topic 可以做到更多容错性,牺牲性能与空间去换取更高的可靠性。

在这里插入图片描述

1.1.4 副本类型:Leader & Follower

假设我们有一份文件,一般情况下,我们对副本的理解应该是有一个正式的完整文件,然后这个文件的备份,我们称之为副本。但是在 Kafka 中,不是这样的,所有的文件都称之为副本,只不过会选择其中的一个文件作为主文件,称之为:Leader(主导)副本,其他的文件作为备份文件,称之为:Fllower(追随)副本。在Kafka中,这里的文件就是分区,每一个分区都可以存在一个或多个副本,只有 Leader 副本才能进行数据的读写,Follower 副本只做备份使用。

在这里插入图片描述

1.1.5 日志:Log

Kafka最开始的应用场景就是日志场景或MQ场景,更多的扮演着一个日志传输和存储系统,所以Kafka 接收到的消息数据最终都是存储在 log 日志文件中的。

主题创建后,会创建对应的数据分区Log日志。并打开文件连接通道,随时准备写入数据。

1.2 创建第一个主题

1.2.1 执行指令
[atguigu@kafka-broker1 ~]$ cd /opt/module/kafka
[atguigu@kafka-broker1 kafka]$ bin/kafka-topics.sh --bootstrap-server kafka-broker1:9092 --create --topic first-topic
1.2.2 ZooKeeper 节点变化

指令执行后,当前 Kafka 会增加一个主题,因为指令中没有配置分区和副本参数,所以当前主题分区数量为默认值 1,编号为 0,副本为 1,编号为所在 broker 的ID值。为了方便集群的管理,创建 topic 时,会同时在 ZK 中增加子节点,记录主题相关配置信息:

/config/topics 节点中会增加 first-topic 节点

在这里插入图片描述

/brokers/topics节点中会增加first-topic节点以及相应的子节点。"

在这里插入图片描述

节点 节点类型 数据名称 数据值 说明
/topics/first-topic 持久类型 removing_replicas
partitions {“0”:[3]} 分区配置
topic_id 随机字符串
adding_replicas
version 3
/topics/first-topic/partitions 持久类型 主题分区节点,每个主题都应该设置分区,保存在该节点
/topics/first-topic/partitions/0 持久类型 主题分区副本节点,因为当前主题只有一个分区,所以编号为0
/topics/first-topic/partitions/0/state 持久类型 controller_epoch 7 主题分区副本状态节点
leader 3 Leader副本所在的broker Id
version 1
leader_epoch 0
isr [3] 副本同步列表,因为当前只有一个副本,所以副本中只有一个副本编号
1.2.3 数据存储位置

主题创建后,需要找到一个用于存储分区数据的位置,根据上面ZooKeeper 存储的节点配置信息可以知道,当前主题的分区数量为1,副本数量为1,那么数据存储的位置就是副本所在的 broker 节点。

[atguigu@kafka-broker3 ~]$ cd /opt/module/kafka/datas
[atguigu@kafka-broker3 datas]$ ll

在这里插入图片描述

[atguigu@kafka-broker3 datas]$ cd first-topic-0
[atguigu@kafka-broker3 first-topic-0]$ ll

在这里插入图片描述

路径中的00000000000000000000.log文件就是真正存储消息数据的文件,文件名称中的0表示当前文件的起始偏移量为0,index文件和timeindex文件都是数据索引文件,用于快速定位数据。只不过index文件采用偏移量的方式进行定位,而timeindex是采用时间戳的方式。

1.3 创建第二个主题

接下来我们创建第二个主题,不过创建时,我们需要设定分区参数 --partitions,参数值为3,表示创建3个分区

1.3.1 执行命令
[atguigu@kafka-broker1 ~]$ cd /opt/module/kafka
[atguigu@kafka-broker1 kafka]$ bin/kafka-topics.sh --bootstrap-server kafka-broker1:9092 --create --topic second-topic --partitions 3
1.3.2 ZooKeeper 节点变化

指令执行后,当前Kafaka会增加一个主题,因为指令中指定了分区数量(–partition 3),所以当前主题分区数量为 3,编号为[0、1、2],副本为1,,编号为所在 broker 的ID值。为了方便集群的管理,创建 Topic 时,会同时在 ZK 中增加子节点,记录主题相关配置信息:

/config/topics 节点中会增加 second-topic 节点。

在这里插入图片描述

/brokers/topics节点中会增加second-topic节点以及相应的子节点。

在这里插入图片描述

节点 节点类型 数据名称 数据值 说明
/topics/second-topic 持久类型 removing_replicas
partitions {“2”:[3],“1”:[2],“0”:[1]} 分区配置
topic_id 随机字符串
adding_replicas
version 3
/topics/second-topic/partitions 持久类型 主题分区节点,每个主题都应该设置分区,保存在该节点
/topics/second-topic/partitions/0 持久类型 主题分区副本节点,因为当前主题有3个分区,第一个分区编号为0
/topics/second-topic/partitions/0/state 持久类型 controller_epoch 7 主题分区副本状态节点
leader 1 Leader副本所在的broker Id
version 1
leader_epoch 0
isr [1] 副本同步列表,因为当前只有一个副本,所以副本中只有一个副本编号
/topics/second-topic/partitions/1 持久类型 主题分区副本节点,因为当前主题有3个分区,当前为第2个分区,所以编号为1
/topics/second-topic/partitions/1/state 持久类型 controller_epoch 7 主题分区副本状态节点
leader 2 Leader副本所在的broker Id
version 1
leader_epoch 0
isr [2] 副本同步列表,因为当前只有一个副本,所以副本中只有一个副本编号
/topics/second-topic/partitions/2 持久类型 主题分区副本节点,因为当前主题有3个分区,当前为第3个分区,所以编号为2
/topics/second-topic/partitions/2/state 持久类型 controller_epoch 7 主题分区副本状态节点
leader 3 Leader副本所在的broker Id
version 1
leader_epoch 0
isr [3] 副本同步列表,因为当前只有一个副本,所以副本中只有一个副本编号
1.3.3 数据存储位置

主题创建后,需要找到一个用于存储分区数据的位置,根据上面ZooKeeper 存储的节点配置信息可以知道,当前主题的分区数量为3,副本数量为1,那么数据存储的位置就是每个分区 Leader 副本所在的 broker 节点。

[atguigu@kafka-broker1 ~]$ cd /opt/module/kafka/datas
[atguigu@kafka-broker1 datas]$ ll

在这里插入图片描述

[atguigu@kafka-broker1 datas]$ cd second-topic-0
[atguigu@kafka-broker1 second-topic-0]$ ll

在这里插入图片描述

1.4 创建第三个主题

接下来我们创建第三个主题,不过创建时,我们需要设定副本参数 --replication-factor,参数值为3,表示每个分区创建3个副本。

[atguigu@kafka-broker1 ~]$ cd /opt/module/kafka
[atguigu@kafka-broker1 kafka]$ bin/kafka-topics.sh --bootstrap-server kafka-broker1:9092 --create --topic third-topic --partitions 3 --replication-factor 3
1.4.1 ZooKeeper 节点变化

指令执行后,当前 Kafka 会增加一个主题,因为指令中指定了分区数量和副本数量(–replication-factor 3),所以当前主题分区数量为3,编号为[0、1、2],副本为3,编号为[1、2、3]。为了方便集群的管理,创建 Topic 时,会同时在 ZK 中增加子节点,记录主题相关配置信息。

/config/topics节点中会增加third-topic节点。

在这里插入图片描述

/brokers/topics节点中会增加third-topic节点以及相应的子节点。x

在这里插入图片描述

节点 节点类型 数据名称 数据值 说明
/topics/third-topic 持久类型 removing_replicas
partitions {“2”:[1,2,3],“1”:[3,1,2],“0”:[2,3,1]} 分区配置
topic_id 随机字符串
adding_replicas
version 3
/topics/third-topic/partitions 持久类型 主题分区节点,每个主题都应该设置分区,保存在该节点
/topics/third-topic/partitions/0 持久类型 主题分区副本节点,因为当前主题有3个分区,第一个分区编号为0
/topics/third-topic/partitions/0/state 持久类型 controller_epoch 7 主题分区副本状态节点
leader 2 Leader副本所在的broker Id
version 1
leader_epoch 0
isr [2,3,1] 副本同步列表,因为当前有3个副本,所以列表中的第一个副本就是Leader副本,其他副本均为follower副本
/topics/third-topic/partitions/1 持久类型 主题分区副本节点,因为当前主题有3个分区,当前为第2个分区,所以编号为1
/topics/third-topic/partitions/1/state 持久类型 controller_epoch 7 主题分区副本状态节点
leader 3 Leader副本所在的broker Id
version 1
leader_epoch 0
isr [3,1,2] 副本同步列表,因为当前有3个副本,所以列表中的第一个副本就是Leader副本,其他副本均为follower副本
/topics/third-topic/partitions/2 持久类型 主题分区副本节点,因为当前主题有3个分区,当前为第3个分区,所以编号为2
/topics/third-topic/partitions/2/state 持久类型 controller_epoch 7 主题分区副本状态节点
leader 1 Leader副本所在的broker Id
version 1
leader_epoch 0
isr [1,2,3] 副本同步列表,因为当前有3个副本,所以列表中的第一个副本就是Leader副本,其他副本均为follower副本

1.5 创建主题流程

1.5.1 命令行提交创建命令

在这里插入图片描述

  1. 通过命令行提交指令,指令中会包含操作类型(–create)、topic的名称(–topic)、主题分区数量(–partitions)、主题分区副本数量(–replication-facotr)、副本分配策略(–replica-assignment)等参数。

  2. 指令会提交到客户端进行处理,客户端获取指令后,会首先对指令参数进行校验。

a. 操作类型取值:create、list、alter、describe、delete,只能存在一个。

b. 分区数量为大于1的整数。

c. 主题是否已经存在

d. 分区副本数量大于1且小于Short.MaxValue,一般取值小于等于Broker数量。

  1. 将参数封装主题对象(NewTopic)。

  2. 创建通信对象,设定请求标记(CREATE_TOPICS),查找Controller,通过通信对象向Controller发起创建主题的网络请求。

1.5.2 Controller 接收创建主题请求

在这里插入图片描述

(1) Controller节点接收到网络请求(Acceptor),并将请求数据封装成请求对象放置在队列(requestQueue)中。

(2) 请求控制器(KafkaRequestHandler)周期性从队列中获取请求对象(BaseRequest)。

(3) 将请求对象转发给请求处理器(KafkaApis),根据请求对象的类型调用创建主题的方法。

1.5.3 创建主题

在这里插入图片描述

(1) 请求处理器(KafkaApis)校验主题参数。

如果分区数量没有设置,那么会采用Kafka启动时加载的配置项:num.partitions(默认值为1)

如果副本数量没有设置,那么会采用Kafka启动时记载的配置项:

default.replication.factor(默认值为1)

(2) 在创建主题时,如果使用了replica-assignment参数,那么就按照指定的方案来进行分区副本的创建;如果没有指定replica-assignment参数,那么就按照Kafka内部逻辑来分配,内部逻辑按照机架信息分为两种策略:【未指定机架信息】和【指定机架信息】。当前课程中采用的是【未指定机架信息】副本分配策略:

分区起始索引设置0

轮询所有分区,计算每一个分区的所有副本位置:

副本起始索引 = (分区编号 + 随机值) % BrokerID列表长度。

其他副本索引 = 。。。随机值(基本算法为使用随机值执行多次模运算)

##################################################################
# 假设 
#     当前分区编号 : 0
#     BrokerID列表 :【1,2,3,4】
#     副本数量 : 4
#     随机值(BrokerID列表长度): 2
#     副本分配间隔随机值(BrokerID列表长度): 2
##################################################################
# 第一个副本索引:(分区编号 + 随机值)% BrokerID列表长度 =(0 + 2)% 4 = 2
# 第一个副本所在BrokerID : 3

# 第二个副本索引(第一个副本索引 + (1 +(副本分配间隔 + 0)% (BrokerID列表长度 - 1))) % BrokerID列表长度 = (2 +(1+(2+0)%3))% 4 = 1
# 第二个副本所在BrokerID:2

# 第三个副本索引:(第一个副本索引 + (1 +(副本分配间隔 + 1)% (BrokerID列表长度 - 1))) % BrokerID列表长度 = (2 +(1+(2+1)%3))% 4 = 3
# 第三个副本所在BrokerID:4

# 第四个副本索引:(第一个副本索引 + (1 +(副本分配间隔 + 2)% (BrokerID列表长度 - 1))) % BrokerID列表长度 = (2 +(1+(2+2)%3))% 4 = 0
# 第四个副本所在BrokerID:1

# 最终分区0的副本所在的Broker节点列表为【3,2,4,1】
# 其他分区采用同样算法

通过索引位置获取副本节点ID

保存分区以及对应的副本ID列表。

(1) 通过ZK客户端在ZK端创建节点:

在 /config/topics节点下,增加当前主题节点,节点类型为持久类型。

在 /brokers/topics节点下,增加当前主题及相关节点,节点类型为持久类型。

(2) Controller节点启动后,会在/brokers/topics节点增加监听器,一旦节点发生变化,会触发相应的功能:

获取需要新增的主题信息

更新当前Controller节点保存的主题状态数据

更新分区状态机的状态为:NewPartition

更新副本状态机的状态:NewReplica

更新分区状态机的状态为:OnlinePartition,从正常的副本列表中的获取第一个作为分区的Leader副本,所有的副本作为分区的同步副本列表,我们称之为ISR( In-Sync Replica)。在ZK路径/brokers/topics/主题名上增加分区节点**/partitions,及状态/state**节点。

更新副本状态机的状态:OnlineReplica

(3) Controller节点向主题的各个分区副本所属Broker节点发送LeaderAndIsrRequest请求,向所有的Broker发送UPDATE_METADATA请求,更新自身的缓存。

Controller向分区所属的Broker发送请求

Broker节点接收到请求后,根据分区状态信息,设定当前的副本为Leader或Follower,并创建底层的数据存储文件目录和空的数据文件。

文件目录名:主题名 + 分区编号

文件名 说明
0000000000000000.log 数据文件,用于存储传输的中心
0000000000000000.index 索引文件,用于定位数据
0000000000000000.timeindex 时间索引文件,用于定位数据

网站公告

今日签到

点亮在社区的每一天
去签到