Kafka

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

消息发送失败

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

危害

幂等性

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

事务机制

消息发送重复

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

MQ简介

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

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

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

MQ优点

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

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

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

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

生产者消费者模式

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

broker

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 独立工作）。

controller

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

message

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

topic

partition

replication

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

producer

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

consumer

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

consumer group

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

offset

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

zookeeper

kafka系统架构

topic在物理层面以partition为分组，一个topic可以分成若干个partition，每个人partition对应于一个文件夹。文件夹命名规则为：topic＋分区序号为防止log文件过大导致数据检索效率低下，将每个partition分为多个segmentsegment文件由三部分组成，分别为“.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数据存储

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

producerInterceptor

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

serializer

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

partitioner

kafka生产者

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

AR、ISR、OSR

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

LEO、HW

集群架构定义

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

leader故障

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

follower故障

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

Unclean Leader Election

副本故障处理

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

acks=0

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

acks=1

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

acks=-1

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

总结

消息确认机制

Kafka