Kafka【基础知识 02】集群+副本机制+数据请求+物理存储+数据存储设计(图片来源于网络)

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: 【2月更文挑战第20天】Kafka【基础知识 02】集群+副本机制+数据请求+物理存储+数据存储设计(图片来源于网络)

1.Kafka集群

Kafka 使用 Zookeeper 来维护集群成员 (Brokers) 的信息。每个 Broker 都有一个唯一标识broker.id,用于标识自己在集群中的身份,可以在配置文件server.properties中进行配置,或者由程序自动生成。下面是 Kafka Brokers 集群自动创建的过程:

  • 每一个 Broker 启动的时候,它会在 Zookeeper 的 /brokers/ids 路径下创建一个临时节点 ,并将自己的 broker.id 写入,从而将自身注册到集群;
  • 当有多个 Broker 时,所有 Broker 会竞争性地在 Zookeeper 上创建 /controller 节点,由于Zookeeper 上的节点不会重复,所以必然只会有一个 Broker 创建成功,此时该 Broker 称为 Controller Broker。它除了具备其他 Broker 的功能外,还负责管理主题分区及其副本的状态。
  • 当 Broker 出现宕机或者主动退出从而导致其持有的 Zookeeper 会话超时时,会触发注册在Zookeeper 上的 Watcher 事件,此时 Kafka 会进行相应的容错处理;如果宕机的是 Controller Broker 时,还会触发新的 Controller 选举。

2.副本机制

为了保证高可用,Kafka 的分区是多副本的,如果一个副本丢失了,那么还可以从其他副本中获取分区数据。但是这要求对应副本的数据必须是完整的,这是 Kafka 数据一致性的基础,所以才需要使用Controller Broker 来进行专门的管理。下面将详解介绍 Kafka 的副本机制。

2.1 分区和副本

Kafka 的主题被分为多个分区 ,分区是 Kafka 最基本的存储单位。每个分区可以有多个副本 (可以在创建主题时使用replication-factor参数进行指定)。其中一个副本是首领副本 (Leader Replica),所有的事件都直接发送给首领副本;其他副本是跟随者副本 (Follower Replica),需要通过复制来保持与首领副本数据一致,当首领副本不可用时,其中一个跟随者副本将成为新首领。

在这里插入图片描述

2.2 ISR机制

每个分区都有一个 ISR(In-Sync Replica) 列表,用于维护所有同步的、可用的副本。首领副本必然是同步副本,而对于跟随者副本来说,它需要满足以下条件才能被认为是同步副本:

  • 与 Zookeeper 之间有一个活跃的会话,即必须定时向 Zookeeper 发送心跳;
  • 在规定的时间内从首领副本那里低延迟地获取过消息。
    如果副本不满足上面条件的话,就会被从 ISR 列表中移除,直到满足条件才会被再次加入。

这里给出一个主题创建的示例:使用--replication-factor指定副本系数为 3,创建成功后使用--describe命令可以看到分区 0 的有 0,1,2 三个副本,只有 0,1 副本都在 ISR 列表中,其中 0 为首领副本。

# 创建一个有3个副本的主题
kafka-topics.sh --create --bootstrap-server tcloud:9092 \
--replication-factor 3 \
--partitions 1 \
--topic test

# 查看主题信息
[root@tcloud ~]# kafka-topics.sh --describe --bootstrap-server tcloud:9092 --topic test
[2021-08-07 18:07:38,587] WARN [AdminClient clientId=adminclient-1] Connection to node 2 (localhost/127.0.0.1:9093) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
Topic: test     PartitionCount: 1       ReplicationFactor: 3    Configs: segment.bytes=1073741824
        Topic: test     Partition: 0    Leader: 0       Replicas: 0,1,2 Isr: 0,1

2.3 不完全的首领选举

对于副本机制,在 Broker 级别有一个可选的配置参数unclean.leader.election.enable,默认值为fasle,代表禁止不完全的首领选举。这是针对当首领副本挂掉且 ISR 中没有其他可用副本时,是否允许某个不完全同步的副本成为首领副本,这可能会导致数据丢失或者数据不一致,在某些对数据一致性要求较高的场景 (如金融领域),这可能无法容忍的,所以其默认值为 false,如果你能够允许部分数据不一致的话,可以配置为 true。

2.4 最少同步副本

ISR 机制的另外一个相关参数是min.insync.replicas, 可以在 Broker 或者主题级别进行配置,代表 ISR 列表中至少要有几个可用副本。这里假设设置为 2,那么当可用副本数量小于该值时,就认为整个分区处于不可用状态。此时客户端再向分区写入数据时候就会抛出异常org.apache.kafka.common.errors.NotEnoughReplicasExceptoin: Messages are rejectedsince there are fewer in-sync replicas than required。

2.5 发送确认

Kafka 在生产者上有一个可选的参数 ack,该参数指定了必须要有多少个分区副本收到消息,生产者才会认为消息写入成功:

  • acks=0 :消息发送出去就认为已经成功了,不会等待任何来自服务器的响应;
  • acks=1 :只要集群的首领节点收到消息,生产者就会收到一个来自服务器成功响应;
  • acks=all :只有当所有参与复制的节点全部收到消息时,生产者才会收到一个来自服务器的成功响应。

3.数据请求

3.1 元数据请求机制

在所有副本中,只有领导副本才能进行消息的读写处理。由于不同分区的领导副本可能在不同的 Broker 上,如果某个 Broker 收到了一个分区请求,但是该分区的领导副本并不在该 Broker 上,那么它就会向客户端返回一个 Not a Leader for Partition 的错误响应。 为了解决这个问题,Kafka 提供了元数据请求机制。

首先集群中的每个 Broker 都会缓存所有主题的分区副本信息,客户端会定期发送发送元数据请求,然后将获取的元数据进行缓存。定时刷新元数据的时间间隔可以通过为客户端配置metadata.max.age.ms来进行指定。有了元数据信息后,客户端就知道了领导副本所在的 Broker,之后直接将读写请求发送给对应的 Broker 即可。

如果在定时请求的时间间隔内发生的分区副本的选举,则意味着原来缓存的信息可能已经过时了,此时还有可能会收到 Not a Leader for Partition 的错误响应,这种情况下客户端会再次求发出元数据请求,然后刷新本地缓存,之后再去正确的 Broker 上执行对应的操作,过程如下图:

在这里插入图片描述

3.2 数据可见性

需要注意的是,并不是所有保存在分区首领上的数据都可以被客户端读取到,为了保证数据一致性,只有被所有同步副本 (ISR 中所有副本) 都保存了的数据才能被客户端读取到。

在这里插入图片描述

3.3 零拷贝

Kafka 所有数据的写入和读取都是通过零拷贝来实现的。传统拷贝与零拷贝的区别如下:

传统模式下的四次拷贝与四次上下文切换

以将磁盘文件通过网络发送为例。传统模式下,一般使用如下伪代码所示的方法先将文件数据读入内存,然后通过 Socket 将内存中的数据发送出去。

buffer = File.read
Socket.send(buffer)

这一过程实际上发生了四次数据拷贝。首先通过系统调用将文件数据读入到内核态 Buffer(DMA 拷贝),然后应用程序将内存态 Buffer 数据读入到用户态 Buffer(CPU 拷贝),接着用户程序通过Socket 发送数据时将用户态 Buffer 数据拷贝到内核态 Buffer(CPU 拷贝),最后通过 DMA 拷贝将数据拷贝到 NIC Buffer。同时,还伴随着四次上下文切换,如下图所示:

在这里插入图片描述

sendfile和transferTo实现零拷贝

Linux 2.4+ 内核通过 sendfile 系统调用,提供了零拷贝。数据通过 DMA 拷贝到内核态 Buffer 后,直接通过 DMA 拷贝到 NIC Buffer,无需 CPU 拷贝。这也是零拷贝这一说法的来源。除了减少数据拷贝外,因为整个读文件到网络发送由一个 sendfile 调用完成,整个过程只有两次上下文切换,因此大大提高了性能。零拷贝过程如下图所示:

在这里插入图片描述
从具体实现来看,Kafka 的数据传输通过 TransportLayer 来完成,其子类PlaintextTransportLayer 的 transferFrom 方法通过调用 Java NIO 中 FileChannel 的transferTo 方法实现零拷贝,如下所示:

@Override
public long transferFrom(FileChannel fileChannel, long position, long count)
throws IOException {
   
   
  return fileChannel.transferTo(position, count, socketChannel);
}

注: transferTo 和 transferFrom 并不保证一定能使用零拷贝。实际上是否能使用零拷贝与操作系统相关,如果操作系统提供 sendfile 这样的零拷贝系统调用,则这两个方法会通过这样的系统调用充分利用零拷贝的优势,否则并不能通过这两个方法本身实现零拷贝。

4.物理存储

4.1 分区分配

在创建主题时,Kafka 会首先决定如何在 Broker 间分配分区副本,它遵循以下原则:

  • 在所有 Broker 上均匀地分配分区副本;
  • 确保分区的每个副本分布在不同的 Broker 上;
  • 如果使用了broker.rack参数为 Broker 指定了机架信息,那么会尽可能的把每个分区的副本分配到不同机架的 Broker 上,以避免一个机架不可用而导致整个分区不可用。

基于以上原因,如果你在一个单节点上创建一个 3 副本的主题,通常会抛出下面的异常:

Error while executing topic command :
org.apache.kafka.common.errors.InvalidReplicationFactor 
Exception: Replication factor: 3 larger than available brokers: 1.

4.2 分区数据保留规则

保留数据是 Kafka 的一个基本特性, 但是 Kafka 不会一直保留数据,也不会等到所有消费者都读取了消息之后才删除消息。相反, Kafka 为每个主题配置了数据保留期限,规定数据被删除之前可以保留多长时间,或者清理数据之前可以保留的数据量大小。分别对应以下四个参数:

  • log.retention.bytes :删除数据前允许的最大数据量;默认值 -1,代表没有限制;
  • log.retention.ms :保存数据文件的毫秒数,如果未设置,则使用 log.retention.minutes 中的值,默认为 null;
  • log.retention.minutes :保留数据文件的分钟数,如果未设置,则使用 log.retention.hours 中的值,默认为 null;
  • log.retention.hours :保留数据文件的小时数,默认值为 168,也就是一周。

因为在一个大文件里查找和删除消息是很费时的,也很容易出错,所以 Kafka 把分区分成若干个片段,当前正在写入数据的片段叫作活跃片段。活动片段永远不会被删除。如果按照默认值保留数据一周,而且每天使用一个新片段,那么你就会看到,在每天使用一个新片段的同时会删除一个最老的片段,所以大部分时间该分区会有 7 个片段存在。

4.3 文件格式

通常保存在磁盘上的数据格式与生产者发送过来消息格式是一样的。 如果生产者发送的是压缩过的消息,那么同一个批次的消息会被压缩在一起,被当作“包装消息”进行发送 (格式如下所示) ,然后保存到磁盘上。之后消费者读取后再自己解压这个包装消息,获取每条消息的具体信息。

在这里插入图片描述

5.数据存储设计

5.1 Partition 的数据文件( Offset,MessageSize,Data )

Partition 中的每条 Message 包含了以下三个属性:Offset,MessageSize,Data,其中 Offset 表示 Message 在这个 Partition 中的偏移量,Offset 不是该 Message 在 Partition 数据文件中的实际存储位置,而是逻辑上一个值,它唯一确定了 Partition 中的一条 Message,可以认为 Offset 是 Partition 中 Message 的 Id;MessageSize 表示消息内容 Data 的大小;Data为 Message 的具体内容。

5.2 数据文件分段 Segment( 顺序读写、分段命令、二分查找 )

Partition 物理上由多个 Segment 文件组成,每个 Segment 大小相等,顺序读写。每个 Segment 数据文件以该段中最小的 Offset 命名,文件扩展名为.log。这样在查找指定 Offset 的 Message 的时候,用二分查找就可以定位到该 Message 在哪个 Segment 数据文件中。

5.3 数据文件索引(分段索引、 稀疏存储 )

Kafka 为每个分段后的数据文件建立了索引文件,文件名与数据文件的名字是一样的,只是文件扩展名为.index。索引文件中并没有为数据文件中的每条 Message 建立索引,而是采用了稀疏存储的方式,每隔一定字节的数据建立一条索引。这样避免了索引文件占用过多的空间,从而可以将索引文件保留在内存中。

在这里插入图片描述

目录
相关文章
|
9天前
|
Kubernetes 应用服务中间件 Docker
Kubernetes学习-集群搭建篇(二) 部署Node服务,启动JNI网络插件
Kubernetes学习-集群搭建篇(二) 部署Node服务,启动JNI网络插件
|
9天前
|
网络安全 数据安全/隐私保护 计算机视觉
2024蓝桥杯网络安全-图片隐写-缺失的数据(0基础也能学会-含代码解释)
2024蓝桥杯网络安全-图片隐写-缺失的数据(0基础也能学会-含代码解释)
|
9天前
|
XML JSON 前端开发
【Flutter前端技术开发专栏】Flutter中的图片、视频与网络资源加载
【4月更文挑战第30天】Flutter是谷歌的开源前端框架,因其高性能、流畅UI和多端运行能力受开发者喜爱。本文聚焦于Flutter中的资源加载:使用`Image`组件加载静态、网络和本地图片;通过`video_player`库加载和播放视频;利用`http`包进行网络资源请求。掌握这些技巧将有助于提升Flutter应用的开发效率和质量。
【Flutter前端技术开发专栏】Flutter中的图片、视频与网络资源加载
|
9天前
|
存储 监控 负载均衡
|
9天前
|
存储 数据采集 NoSQL
使用Python打造爬虫程序之数据存储与持久化:从网络到硬盘的无缝对接
【4月更文挑战第19天】本文探讨了爬虫中的数据存储与持久化技术,包括文本文件存储、数据库(关系型与非关系型)、NoSQL数据库和键值存储,以及ORM框架的使用。根据数据类型、规模和访问需求选择合适存储方式,并注意数据安全、备份和恢复策略。正确选择和应用这些技术能有效管理和利用爬取数据。
|
9天前
|
SQL 存储 分布式计算
Hive【基础 01】核心概念+体系架构+数据类型+内容格式+存储格式+内外部表(部分图片来源于网络)
【4月更文挑战第6天】Hive【基础 01】核心概念+体系架构+数据类型+内容格式+存储格式+内外部表(部分图片来源于网络)
37 1
|
9天前
|
分布式计算 资源调度 Hadoop
Hadoop【基础知识 03+04】【Hadoop集群资源管理器yarn】(图片来源于网络)(hadoop fs + hadoop dfs + hdfs dfs 使用举例)
【4月更文挑战第5天】Hadoop【基础知识 03】【Hadoop集群资源管理器yarn】(图片来源于网络)Hadoop【基础知识 04】【HDFS常用shell命令】(hadoop fs + hadoop dfs + hdfs dfs 使用举例)
64 9
|
9天前
|
分布式计算 资源调度 Hadoop
Hadoop【基础知识 03】【Hadoop集群资源管理器yarn】(图片来源于网络)
【4月更文挑战第4天】Hadoop【基础知识 03】【Hadoop集群资源管理器yarn】(图片来源于网络)
38 4
|
9天前
|
消息中间件 存储 算法
深入了解Kafka的数据持久化机制
深入了解Kafka的数据持久化机制
52 0
|
7天前
|
消息中间件 关系型数据库 Kafka
实时计算 Flink版产品使用合集之想要加快消费 Kafka 数据的速度,该怎么配置参数
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
19 2

热门文章

最新文章