首页  思维导图  详情

Kafka

2023-02-23 21:55:02   0  举报





AI智能生成

详细记录了kafka0.11版本的知识点。与市面上公司使用的稳定版本所契合，实用性高

学霸笔记

学习笔记

系统架构

大数据

作者其他创作

大纲/内容

消息发送失败

消息不是一直都会发送成功的，也可能发送失败。发送失败分为可重试恢复错误和不可重试恢复错误。 可重试恢复错误：找不到leader；找不到目标分区，这种情况往往重试一下就能发送成功。 不可重试恢复错误：消息体过大、缓冲区满了。这种情况重试也会失败，因为消息体过大除非减少消息量，或者采用压缩，重试无用。

消息发送重复

危害

可能会导致消息被重复处理，例如银行，危害会很大

幂等性

为了实现producer的幂等语义，kafka引入了producer ID和sequence Number， 每个新的producer在初始化的时候会被分配一个唯一的PID，该PID对用户完全透明而不会暴露给用户。 总结下来，重复数据的判断标准为：具有<PID,partition,seqNumber>相同主键的消息提交时，broker只会持久化一条。 其中PID时kafka每次重启都会分配一个新的；partition表示分区号；seqNumber是单调自增的， 所以幂等性只能保证在单分区单会话内不重复。

事务机制

与幂等性有关的另外一个特性就是事务，前面我们提到幂等性只能保证在单分区单会话内不重复， 所以幂等性不能跨多个分区运作，而事务可以弥补这个缺陷。事务可以保证对多个分区写入操作的原子性。 操作的原子性是指多个操作要么全部成功，要么全部失败，不存在部分成功、部分失败的可能。

MQ简介

消息队列是一种进程间通信或同一进程的不同线程间的通信方式，是基础数据结构中“先进先出”的一种数据结构。

MQ优点

解耦：将消息写入消息队列，需要消息的系统自己从消息队列中订阅，系统不需要做任何修改即可实现业务，降低模块间的耦合

异步：将消息写入消息队列，非必要的业务逻辑（短信、邮件）以异步的方式运行，加快响应速度，提高用户体验。

削峰：并发量激增的时候，所有的请求先写入消息队列，然后系统按照存储服务能处理的并发量，从消息队列中慢慢拉取消息， 提供高峰期业务处理能力，避免系统瘫痪。在生产中，这个短暂的高峰期积压是允许的。

生产者消费者模式

支持解耦：假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法，那么生产者对于消费者就会产生依赖。

支持并发：生产者直接调用消费者的某个方法过程中函数调用时同步的                  万一消费者处理数据很慢，生产者就会白白糟蹋大好时光

支持忙闲不均：缓冲区还有另一个好处，如果制造数据的速度时快时慢，缓冲区的好处就体现出来了。                         当数据制造快的时候，消费者来不及处理，未处理的数据可以暂时存在缓冲区中。                         等生产者的制造速度慢下来，消费者再慢慢处理掉。

数据传递流程：生产者消费者模式，即N个线程进行生产，同时N个线程进行消费，两种角色通过内存缓冲区进行通信。                         生产者负责向缓冲区里面添加数据单元                         消费者负责从缓冲区里面取出数据单元。一般遵循先进先出原则。

kafka系统架构

broker

每个kafka server称为一个Broker，多个borker组成 Kafka Cluster。 一个broker可以维护多个topic

controller

Controller 表示控制器，Kafka 节点中的主节点。集群中任意一台 Broker 都能充当控制器的角 色，但是，在运行过程中，只能有一个 Broker 成为控制器，行使其管理和协调的职责。 在分布式系统中，通常需要有一个协调者，该协调者会在分布式系统发生异常时发挥特殊的作用。 在 Kafka 中该协调者称之为控制器（Controller），其实该控制器并没有什么特殊之处，它本身也 是一个普通的 Broker，只不过需要负责一些额外的工作： Broker 管理（新增 Broker、Broker 主动关闭、Broker 故障）； Topic 管理（创建主题、删除主题）； Partition 管理（Leader 分区选举、增加分区、Rebalance 分区）。 值得注意的是：Kafka 集群中始终只有一个 Controller Broker。 2.8 版本以前通过 ZooKeeper 实现选主 每个 Broker 启动时都会尝试在 ZooKeeper 上注册 /controller 临时节点来竞选控制 器，第一个创建 /controller 节点的 Broker 会被指定为控制器。竞争失败的节点也会 依赖 Watcher 机制，监听这个节点，如果控制器宕机了，那么其它 Broker 会继续争 抢，实现控制器的 Failover。 此处可以想想NameNode如何竞选主备节点的 2.8 版本以后新增了 KRaft 实现选主（抛弃 ZooKeeper 独立工作）。

message

Message 表示消息。通过 Kafka 集群进行传递的消息对象实体，存储需要传送的信息。 Message 是实际发送和订阅的信息的实际载体，Producer 发送到 Kafka 集群中的每条消息，都被 Kafka 包装成了一个个 Message 对象，之后再存储在磁盘中，而不是直接存储的。 其中 key 和 value 存储的是实际的 Message 内容，长度不固定，而其他都是对 Message 内 容的统计和描述，长度固定。因此在查找实际 Message 过程中，磁盘指针会根据 Message 的 offset 和 message length 计算移动位数，以加速 Message 的查找过程。之所以可以这 样加速，因为 Kafka 的 .log 文件都是顺序写的，往磁盘上写数据时，就是追加数据，没有随 机写的操作。

topic

每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic，类似于数据库的table或者ES 的Index。 逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可（实际上 生产或消费数据不必关心数据存于何处） Producer将消息推送到topic，由订阅该topic的Consumer从topic中拉取消息。

partition

Kafka 中 Topic 被分成多个 Partition 分区，每个topic至少有一个partition。多分区可以提高 Topic的 Topic 是一个逻辑概念，Partition 是最小的存储单元吞吐量和并行度，拥有着一个 Topic 的部分数 据。 每个 Partition 都是一个单独的 log 文件，每条记录都以追加的形式写入。 当生产者产生数据的时候，根据分配策略选择分区,然后将消息追加到指定的分区的末尾（队列） 每条消息都会有一个自增的编号（Offset偏移量） 它表示分区中每条消息的位置信息，是一个单调递增且不变的值。换句话说，offset可以用来 唯一的标识分区中每一条记录。 每个Partition都有自己独立的编号 partition中的数据是有序的，不同partition间的数据是无序的。 如果topic有多个partition，消费数据时就不能保证数据的顺序。严格保证消息的消费顺序的场景 下，需要将partition数目设为1。

replication

数据会存放到topic的partation中，partation会存放到Broker上，但是有可能会因为Broker损坏 导致数据的丢失 Kafka为一个Partition生成多个副本，并且把它们分散在不同的Broker上（就是说需要对数据进行 备份，备份多少取决于你对数据的重视程度）。 如果一个Broker故障了，Consumer可以在其他Broker上找到Partition的副本，继续获取消息。 我们将分区的分为Leader(1)和Follower(N-1),备份数设置为N，表示主+备=N(参考HDFS) Leader负责写入和读取数据 Follower只负责备份 保证了数据的一致性 Leader 每个partition有多个副本，其中有且仅有一个作为Leader，Leader是当前负责数据的读写的 partition。 Follower Follower跟随Leader，所有写请求都通过Leader路由，数据变更会广播给所有Follower， Follower与Leader保持数据同步。 如果Leader失效，则从Follower中选举出一个新的Leader。

producer

生产者即数据的发布者，该角色将消息发布到Kafka的topic中。 broker接收到生产者发送的消息后，broker将该消息追加到当前用于追加数据的segment文件 中。 生产者发送的消息，存储到一个partition中，生产者也可以指定数据存储的partition。 比如我们经常去淘宝购物，当你打开淘宝的那一刻，你的登陆信息，登陆次数都会作为消息 传输到 Kafka 后台，当你浏览购物的时候，你的浏览信息，你的搜索指数，你的购物喜好都 会作为一个个消息传递给 Kafka 后台，然后淘宝会根据你的喜好做智能推荐。

consumer

Kafka 不像普通消息队列具有发布/订阅功能，Kafka 不会向 Consumer 推送消息。 Consumer 必须自己从 Topic 的 Partition 拉取消息。Consumer 连接到一个 Broker 的 Partition，根据偏移量依次读取消息。 消息的 Offset 就是 Consumer 的游标，根据 Offset 来记录消息的消费情况。 读完一条消息之后，Consumer 会推进到 Partition 中的下一个 Offset，继续读取消息。

consumer group

每个Consumer属于一个特定的Consumer Group（需要为每个Consumer指定Group ID）。 定义消费者组为了将多个消费者集中到一起去处理某一个Topic的数据，可以提高数据的消费能力 整个消费者组共享一组偏移量(因为一个Topic有多个分区，每个分区都有自己的偏移量)，防止数 据被重复读取 后续将具体讲解消费数据的分区分配策略。

offset

消息写入的时候，每一个分区都有一个 offset，这个 offset 就是生产者的 offset，同时也是这个 分区的最新最大的 offset。 例如生产者写入的 offset 最新最大的值是 12，而当 Consumer A 进行消费时，从 0 开始消费，一 直消费到了 9，消费者 A 的 offset 就记录在 9。等它下一次再来消费时，它可以选择接着上一次 的位置消费，也可以选择从头消费，或者跳到最近的记录并从“现在”开始消费 偏移量可以唯一的标识一条消息，且可以决定读取数据的位置，消费者通过偏移量来决定下次读取 的消息。 我们某一个业务也可以通过修改偏移量达到重新读取消息的目的，偏移量由用户控制。这其中不会 有线程安全的问题，因为消息被消费之后，并不会被马上删除。这样还可以让多个业务重复使用 Kafka 的消息，但是消息最终还是会被删除，默认生命周期为 1 周（7 * 24小时）。

zookeeper

Kafka 集群使用 ZooKeeper 来负责集群元数据的管理、控制器的选举等操作。 在每个 Broker 启动的时候，都会和 ZooKeeper 进行交互，这样 ZooKeeper 就存储了集群中所有 的主题、配置、副本等信息，还有一些选举、扩容等机制也都依赖 ZooKeeper Zookeeper上常见的节点说明： Broker注册并监控状态 znode:/brokers/ids，保存了所有 Broker id，实现对 Broker 的动态监控。 Topic注册 znode:/brokers/topics，保存了所有 Topic。 生产者负载均衡 每个Broker启动时，都会完成Broker注册过程，生产者会通过该节点的变化来动态地感 知到Broker服务器列表的变更 offset维护

kafka数据存储

文件存储结构

topic在物理层面以partition为分组，一个topic可以分成若干个partition，每个人partition对应于一个文件夹。 文件夹命名规则为：topic＋分区序号 为防止log文件过大导致数据检索效率低下，将每个partition分为多个segment segment文件由三部分组成，分别为“.index”文件、“.log文件”、“timeindex”文件 partition全局的第一个segment从0开始，后续每个segment文件名为当前segment文件第一条消息的offset值。 segment文件的存储优点： 加快查询效率：通过将分区的数据根据offset划分到多个比较小的segment文件，在检索数据时，可以根据offset快速定位数据所在的segment 加载单个segment文件查询数据，可以提高查询效率。 删除数据时减少io：删除数据时，kafka以segment为单位删除某个segment的数据，避免一条一条的删除，增加io负担，性能较差。

索引与数据

索引分为两类：一种时稠密索引，一种时稀疏索引。 稠密索引：即每一条记录对应一个索引字段，访问速度快，但是维护成本大。 稀疏索引：将记录分为若干个片段，为每个片段建立一条索引字段。 稀疏索引字段，要求索引字段是按顺序排序的，否则无法有效索引。kafka选择的就是稀疏索引。 默认情况下，.log文件每增加4096个字节，在.index中增加一条索引。4k

数据检索流程

1：根据offset计算这条offset是这个文件中的第几条。 2：读取.index索引文件，根据二分检索，从索引中找到离这条数据最近偏小的位置。 3：读取.log文件从最近位置读取到要查找的数据

kafka生产者

producerInterceptor

拦截器功能最早在kafka0.10.0.0中引入，kafka一共有两种拦截器 生产者拦截器和消费者拦截器。生产者拦截器可以用来在消息发送前做一些处理

serializer

生产者需要使用序列化将对象转换成字节数组才能够通过网络发送给kafka。 消费者端需要使用反序列化把从kafka接收到的字节数组转换成相应的对象。

partitioner

分区选择器，默认是对于key进行hash计算然后对于总分区数求模以此得到被发送的分区号， 当然我们实现producer时可以自定义partition，或者指定特定分区。 分区的作用就是为消息分配消息。单个分区的消息是有序的。

集群架构定义

AR、ISR、OSR

kafka的数据是多副本的，每个partition都会有N个副本。每个topic下的每个分区下都有一个leader和（N-1）个follower 每个follower的数据都是同步leader的，follower主动拉取leader的数据 leader负责指定分区所有读写操作，follower复制指定分区日志备份和leader失效后的选举。 replication的个数应小于broker的个数 AR：用来标识副本的全集 OSR:离开同步队列的副本 ISR：加入同步队列的副本 ISR=leader+没有落后太多的副本 AR=OSR+ISR ISR判定标准：默认10s。isr中的follower没有向leader发送心跳包就会被移除。

LEO、HW

LEO：日志末端偏移量，代表日志文件中下一条待写入消息的offset，这个offset上实际是没有消息的。 HW：副本的高水印值，取值于所有副本中最小的LEO值。所以仅HW之前数据的消费者可见。

副本故障处理

leader故障

leader发生故障后，优先从isr中选出一个新的leader 为保证多个副本之间的数据一致性，其余的follower会先将各自的log文件高于HW的部分截掉，然后从新的leader同步数据 这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。

follower故障

follower发生故障后被临时提出isr 这个期间leader和其它follower继续接收数据，也有可能发生leader选举等事件 待该follower恢复后，follower会读取本地磁盘记录上的HW，并将日志文件大于等于HW的部分截取掉，从HW重新向leader进行同步 等该follower的LEO大于等于该partition的HW，即follower追上leader之后，就可以重新加入isr了

Unclean Leader Election

ISR 为空。因为 Leader 副本天然 就在 ISR 中，如果 ISR 为空了，就说明 Leader 副本也“挂掉”了，Kafka 需要重新选举一个新的 Leader。 开启 Unclean 领导者选举可能会造成数据丢失，但好处是，它使得分区 Leader 副本一直存在， 不至于停止对外提供服务，因此提升了高可用性。反之，禁止 Unclean 领导者选举的好处在于维 护了数据的一致性，避免了消息丢失，但牺牲了高可用性。

消息确认机制

acks=0

这个条件下，producer无需等待来自leader的确认而继续发送下一批消息。 当broker故障时有可能丢失数据。

acks=1

producer在isr中的leader已成功收到的数据并得到确认后发送下一条message 如果在follower同步成功之前leader故障，那么将会丢失数据。

acks=-1

producer需要等待isr中的所有follower都确认收到数据后才算一次发送完成，可靠性最高。 在broker发送ack时，leader发生故障，则会造成数据重复

总结

acks=0，生产者发送数据之后就不管了，可靠性差，效率高； acks=1，生产者发送数据后leader应答，可靠性中等，效率中等； acks=all，生产者发送数据后leader和isr队列里面所有follower应答，可靠性搞，效率低。