# Kafka ## 简介 ### 消息队列应用场景 缓存/削峰、解耦、异步通信 ### 缓存/削峰 当生产端生产数据的速率大于消费端消费数据的速率时，消息队列可用于对消费不完的数据进行缓存 ### 解耦 当数据生产方来源存在多个（例如数据库、网络接口等），且消费方也存在多种（Hadoop大数据平台、spring boot服务实例等）时，可以将消息队列中存储的消息作为数据从生产端到消费端的中间格式。 生产端负责将产生的数据转化成特定格式的消息发送到消息队列，而消费端负责消费消息队列中的消息。不同种类的生产端和消费端只用针对消息的队列中的消息各自进行适配即可。 ### 异步通信 通过消息队列，可以实现多个服务实例之间的异步调用，服务调用方在将调用信息封装到消息并发送消息到mq后，即可返回，并不需要同步等待。被调用方可以异步的从mq中获取消息并进行消费。 ### 消息队列模式 #### 点对点 一个消息队列可以对应多个生产者和多个消费者，**但消息只能被一个消费者进行消费**，消费者将数据拉取并消费后，向mq发送确认，mq中将被消费的消息删除 #### 发布订阅模式 **生产者产生的数据可以被多个消费者消费**，生产者将消息发送到topic，订阅该topic的消费者会拉取消息进行消费，消费完成后并不会删除该消息，该消息仍可被其他订阅该topic的消费者进行消费。（消息并不会永久保存在消息队列中，通常会设置超期时间，消息保存超过该时间后自动删除） ### kafka架构 #### 分区存储 kafka将一个topic分为了多个分区，用于分布式存储数据。而每个分区都跨多台服务器实例进行存储，从而增加容错性。对特定分区来说，其存在于多台服务器实例，其中一台为主机，其他服务器为从机，**主机会处理对该分区数据所有的读写请求**（由于读写操作全都在主机上发生，而每台服务器都担当了某些分区的主机和其他分区的从机，故而读写请求在不同的服务器之间进行了负载均衡），而从机只会被动的复制主机修改。 #### group消费 每个消费者实例都属于一个consume group，一个consume group则是可以保存多个消费者实例。对于发送到kafka mq topic的每一条消息，都会被广播给订阅了该topic的每个consume group。而consume group接收到消息后，只会将消息发送给group中的一个消费者实例来进行处理，故而在consume group中，实现了消息在不同消费实例之间的负载均衡。 > 在kafka mq中，topic的实际订阅者是consume group，kafka mq会将topic中的消息广播给所有订阅了该topic的consume group，而每条消息都只会被consume group中的一个消费者实例进行消费 > > 在一个consume group中，每个消费者实例都负责某一topic中相应的分区，每个特定分区只有一个消费者实例负责。 #### zookeeper zookeeper作为注册中心，用于记录存在多个kafka实例时，当前已上线且状态正常的kafka实例，以及各个分区leader实例的信息 #### 生产者 ##### 分区 生产者客户端将会决定将消息发送到某个分区，可以通过负载均衡随机决定将消息发送到哪个分区，也可以提供一个key并通过算法决定将消息发送到哪个分区。 决定将消息发送到哪个分区后，生产者会直接将消息发送到该分区对应的leader broker中。 ##### 批量发送 生产者并不会每次产生消息后都立即将消息发送到broker，而是会累积消息，**直到消息数据积累到特定大小（默认为16K）后**，才会将消息发送给broker。 > **消息发送条件** > > - batch.size: 当消息累计到特定大小（默认为16K）后，发送给broker。如果消息的大小大于`batch.size`,那么并不会对消息做累积操作。**发送到broker的请求将会包含多个batch，每个batch对应一个分区，不同分区的消息通过不同的batch进行发送** > - linger.ms: 生产者将会把发送给broker的两个请求之间的消息都累积到一个batch里，该batch将会在下次发送请求给broker时发送。但是，对特定分区累积大小如果没有达到`batch.size`的限制，哪个通过linger.ms来控制该消息延迟发送的最长时间。`linger.ms`单位为ms，**其默认值为0，代表即使消息累积没有达到`batch.size`，也会立马发送给broker**。若`linger.ms`设置为1000，则代表没有累积到`batch.size`大小的消息也会在延迟1s后发送给broker ##### ack级别 生产者将消息发送到broker后，broker对生产者有三种应答级别： - 0：生产者发送数据后无需等待broker返回ack应答 - 1：生产者发送消息到broker后，leader broker将消息持久化后向生产者返回ack - -1：生产者发送过来的数据，leader broker和isr队列中所有的broker都持久化完数据后，返回ack ##### 异步发送 kafka生产者异步发送是指生产者调用kafka客户端接口将发送的消息传递给kafka的客户端，此时消息并未发送到broker，而异步调用接口即返回。调用异步接口可以通过指定回调来获取消息传递到的topic、分区等信息。 ##### 同步发送 除了异步调用接口外，还可以调用同步调用的接口来发送消息。在调用异步接口发送消息时，仅仅将消息放到缓冲区后调用就返回，可能放到缓冲区的消息并没有发送到broker。可以通过调用同步发送消息的接口来发送消息，其会阻塞并等待broker将消息持久化后返回的异常或者ack应答。 想要同步发送消息，只需要对异步接口返回的future对象调用`.get`即可，其会等待future对象完成。 ##### 生产者发送消息的分区策略 在生产者发送消息时，需要决定将消息发送到哪个分区，决定策略如下所示： 1. 如果发送消息时指定了要发送到哪个分区，那么发送到哪个分区 2. 如果发送消息时没有指定分区，但是指定了消息的key，那么对key进行散列，根据key散列后的值决定发送到哪个分区 3. 如果发送消息时没有指定发送到哪个分区，也没有指定key，那么根据黏性策略发送消息 > 黏性分区策略 > > 黏性分区策略会随机选择一个分区进行发送，并尽可能的一直使用该分区，直到该分区的batch size已满，此时kafka会随机再选择另一个分区进行发送 除了上述分区策略外，kafka支持自定义分区策略。 ##### kafka消息传递语义 1. 至少一次：acks=-1情况下，消息能被确保传递不丢失，但是可能会存在消息重复传递的问题 2. 至多一次：acks=0的情况下，kafka能保证消息最多只传递一次的问题，但是无法保证消息会发送到broker，可能存在消息丢失问题 3. 精确一次：数据既不会重复也不会丢失 为了保证精确一次，kafka引入了两个特性：`幂等性`和`事务` > 幂等性 > > 生产者在初始化时会被分配一个producer id，并且生产者在发送消息时，针对每条消息都存在序列号。生产者每次发送消息时，都会附带pid（生产者id）和消息的序列号。序列号是针对topic分区的，生产者向分区发送消息时，针对每个分区都维护了一套序列号。当消息被发送给broker后，其会比较当前分区来源PID和消息PID相同，并且已经被ack的最大序列号，如果当前消息的序列号小于已经ack的序列号，说明该消息已经发送过，属于重复消息，被丢弃。 > > **由于每个新的生产者实例都会被分配一个新的producer id，那么只能在单个生产者会话中保证幂等。** > > 生产者可以配置`enable.idempotence`属性用于控制幂等是否开启，默认情况下该属性值为true，默认开启幂等性，生产者会保证对每条消息只有一个副本被写入 > 事务 > #### 概念 > kafka事务能够保证应用能够原子的向多个topic分区中写入数据，所有写入操作要么同时成功，要么同时失败。 > > 另外，消费过程可以看作是对offset topic的写操作，那么kafka事务允许将消息的消费和消息的生成包含在一个原子的单元中。 > > #### 保证 > 为了实现kafka事务，需要应用提供一个全局唯一的transactionId，该transactionId会稳定贯穿所有该应用的会话。当提供了transactionId后，kafka会做出如下保证： > > - 只有一个活跃的生产者具有指定transactionId。为了实现该保证，当具有相同transactionId的新生产者实例上线后，会隔离旧的实例 > - 在应用会话之间将会进行事务的恢复。如果应用实例宕机，那么那么下一个实例恢复工作前将会有一个干净的状态，任何尚未完成的事务都会处于完成状态（提交或异常退出） > > #### 事务协调器 > 每个生产者实例都被分配了一个事务协调器，分配producer id、管理事务等逻辑都由事务协调器负责 > #### Transaction Log > Transaction Log是一个内部的kafka topic，类似于consumer offset topic，transaction log是一条持久化和主从复制的记录，记录了每个事务的状态。transaction log是对事务协调器的状态存储，最新版本log快照封装了当前每个活跃事务的状态。 > #### 控制消息 > 控制消息是写入用户topic中的特殊消息，由客户端进行处理，但是不暴露给用户。例如，其被用于broker告知消费者其先前拉取的消息是否已经被原子的提交。 > #### transactionId > transactionlId用于唯一标识producer，并且transactionId是持久化的（producer id在生产者实例重启后会重新获取）。如果不同生产者实例具有相同的transactionId，那么会恢复或终止上一生产者实例所初始化的任何事务 > > #### 数据流 > ##### 发现事务协调器 > 由于事务协调器是分配PID和管理事务的中心，首先生产者会发送请求给任一broker来询问其所属的事务协调器 > ##### 获取PID > 查询到其所属事务协调器后，会获取producer id。如果transactionId已经被指定，那么在获取PID的请求中会包含transactionId，并且transactionId和PID的关联将会被记录到transaction log中,以便于之后如果上线具有相同transactionId的新实例，能够根据transactionId该映射返回旧实例的PID。 > > 在上线新实例，返回具有相同transactionId的旧实例后，会执行如下操作： > > 1. 隔离旧实例，旧实例所属的事务无法再继续推进 > 2. 针对旧生产者实例所属的尚未完成的事务进行恢复操作 > > 如果请求PID时transactionId尚未被指定，那么PID将会被分配，但是生产者实例只能在单个会话中使用幂等语义和事务语义 > > #### 开启事务 > 可以通过beginTransaction()接口调用来开启事务，生产者会将本地状态改为事务已开启，但是事务协调器直到发送了第一条记录时才会将事务状态标记为已开启 > #### 消费-处理-生产 > 开启事务之后，生产者可以对消息进行消费和处理，并且产生新的消息， > ##### AddPartitionsToTxnRequest > 当某个topic分区第一次作为事务的一部分被执行写操作时，生产者将会发送AddPartitionsToTxnRequest请求到事务协调器。将该topic分区添加到事务的操作将会被事务协调器记录。可以通过该记录信息来对事务中的分区进行提交或回滚操作。如果第一个分区被添加到事务中，事务协调器将会开启事务计时器。 > ##### ProduceRequest > 生产者会向topic分区发送多个ProduceRequest来写入消息，请求中包含PID、epoch、序列号。 > > #### 提交或回滚事务 > 一旦消息被写入，用户必须调用 commitTransaction或abortTransaction方法来对事务进行提交或者回滚。 > ##### 数据乱序 kafka如果想要保证消息在分区内是有序的，那么需要开启幂等性（默认开启）并且`max.inflight.requests.per.connection`需要设置小于或等于5（默认情况下该值为5），在不开启幂等性的情况下如果想要保证分区内数据有序需要设置`max.inflight.requests.per.connection`的值为1. > `max.inflight.requests.per.connect` > > 该值用于设置客户端向broker发送数据时，在阻塞前每个连接能发送的未收到ack的请求数目。如果该值被设置为大于1并且幂等性未开启，那么在消息发送重试时可能会导致消息的乱序，此时需要开启幂等性或者关闭消息重试机制才能重新保证幂等性。 > > 并且，启用幂等性需要该值小于或等于5，如果该值大于5，且幂等性开启 > - 如果幂等性未显式指定开启，那么在检测到冲突设置后，幂等性会关闭 > - 如果显式指定幂等性开启并且该值大于5，那么会抛出ConfigException异常 #### broker ##### broker选举 每个broker节点中都存在controller模块，只有ISR中的节点才能够参与选举（acks=all时，也只需要ISR队列中所有节点都同步消息，才将消息视为已提交）。controller会通过watch来监听zookeeper中节点信息的变化，如果某个分区的leader broker宕机，那么controller模块在监听到变化后会重新开始选举，在ISR中重新选出一个节点作为leader。 ##### partition reassign 如果kafka集群在运行时新增了新的节点，**此时节点中旧的topic分区并不会自动同步到新增节点中，如果要将旧topic分区存储到新增broker节点，可以调用重新分区的命令。** 可以通过`kafka-reassign-partitions.sh --generate`命令对分区在所有节点（包含新节点）之间进行重新分配，此命令调用后会生成一个topic分区在节点间重新分配的计划。 在确认了生成的重新分配计划后，可以调用`kafka-reassign-partitions.sh --execute`命令来执行新生成的重新分配计划。 如果想要在kafka集群中退役一台broker server，不能直接关闭该broker节点，也需要跟新增broker节点类似，重新生成一个分区分配计划，分配计划中移除待下线节点，然后执行该分配计划。执行分配计划后，该节点上便不再存储分区，此后该节点并可以安全的关闭。 ##### broker宕机对分区造成的影响 broker宕机并不会导致分区的重新分配，例如一个分区的replica-factor为2，存在broker-1，broker-2，broker-3，分区存储在(1,3)服务器上，leader为1号服务器。如果broker-1宕机，那么分区的leader会自动切换为broker-3,但是分区存在的副本数量从2个变成了一个，ISR中也只有(3),此时并不会在未存储该分区的broker-2节点上新增一个副本。 如果宕机的broker-1重新再上线，那么broker-1会重新存储之前负责存储的分区，但是此时分区对应的leader节点仍然是broker-2，此时broker-2作为分区的leader会导致读写操作全都转移到broker-2节点，负载均衡会造成偏移。 > `auto.leader.rebalance.enable` > > 该属性默认设置为true，一个后台线程会定期(时间间隔为`leader.imbalance.check.interval.seconds`,默认为300s)检测分区leader是否为默认的perferred leader。如果分区leader不为preferred leader的数量超过一定的比率(` leader.imbalance.per.broker.percentage`，默认为10%)，会触发将分区leader改为默认preferred leader的操作。 ##### broker AR & OSR kafka中副本的默认数量为1个，通常生产环境将其配置为2个或2个以上，以便在leader副本宕机后follower副本能够继续服务，避免数据丢失。 - AR: AR代表集群中分区对应的所有副本集合，all-replicas - ISR： 代表集群中处于同步状态的副本集合，in-sync-replicas 如果ISR中的节点长期未从leader中同步数据，会被剔除出ISR，最长未同步的时间由`replica.lag.time.max.ms`控制，默认为30s。如果leader对应的broker宕机，那么新leader将会从ISR中选举产生 - OSR：代表集群中不处于同步状态的副本集合，为AR - ISR ##### leader选举机制 leader选举根据AR中节点的排序来决定，能够成为leader的节点需要再ISR中存活，然后ISR中存活节点在AR中排序最靠前的将会被选举为leader。如果leader发生宕机，那么由ISR中存活且AR中排序最靠前的的节点成为新leader。 > 选举机制中leader决定是按照AR中broker节点的顺序进行决定的，每一个分区都有一个默认的preferred leader。如果某些节点宕机后再恢复，ISR中节点的顺序将会发生变化，但是AR中的节点顺序并不变，并且preferred leader也不会发生变化。 ##### follower故障恢复细节 follower故障恢复细节中，涉及到如下两个概念： - LEO(Log End Offset)：每个副本结束offset的下一个位置，类似于java中的List.size() - HW(High WaterMark)：当前ISR队列中所有副本最小的HOW（水位线，类似于木桶效应中最短的那一根木板） 如果follower发生故障，那么会按顺序发生如下： 1. 故障follower会被剔除出ISR 2. follower故障期间ISR中剩余的follower节点和leader节点会继续接受数据 3. 故障follower如果恢复，会读取宕机前记录的旧集群HW，并且将大于等于该旧HW的log记录全部都删除，并且从leader重新同步记录 4. 当故障恢复的follower从旧HW位置同步消息到当前集群中的新HW位置时，此时故障恢复的log区数据已经同步到新HW的水位线水平，此时故障恢复的follower节点可以重新加入到ISR中 ##### leader宕机的细节 如果leader broker发生宕机，那么新选举成为leader的follower将会成为leader，并且所有节点会将大于等于HW的数据丢弃(由于宕机的是leader，LEO最大，故而leader宕机不会对HW造成影响)，并且重新从新选举的leader中同步数据。 由上可知，如果leader宕机前，其他follower尚未同步完leader中全部的消息，那么leader宕机后可能会发生消息的丢失。如果需要确保消息不丢失，需要设置生产者的acks为all，确保消息再提交前同步到ISR中所有的节点中。 ##### kafka分区存储机制 kafka中topic是一个逻辑概念，topic由分区组成，每个分区则是可以看作一个log文件，log中存放生产者产生的数据。生产者产生的消息会被追加到log文件的末端，由于是线性追加，不涉及到平衡树等数据结构，故而**kafka追加消息时，不管当前分区数据存储量大小，追加数据的开销都是相同的，追加操作不会随着数据量变大而变慢**。 为了防止分区对应的log文件过大而导致的数据定位效率低下，kafka采用了分片和索引的机制，将每个分区分为了多个segment。单个segment默认存储的数据量的大小为1G，且单个segment由如下文件组成： - .log文件：日志文件，用于存储消息（log文件的命名以当前segment中第一条消息再分区中的offset来命名） - .index：偏移量索引文件，用于存储偏移量和position的映射关系，用于快速定位消息 - .timeindex：时间戳索引文件，该文件用于存储消息对应的时间戳信息，kafka中的消息默认保存7天后丢弃，通过时间戳信息来决定消息是否应该丢弃 > index索引 > > kafka中的索引为稀疏索引，默认每往log中写入4KB的数据，.index文件中会记录一条偏移量索引信息。 > > 可以通过`log.index.interval.bytes`来配置索引记录的密度，每写入多少数据才记录一条索引 ##### kafka文件清除策略 kafka中默认消息保存的时间为7天，可以通过修改如下配置来对默认保存时间进行修改： - log.retention.hours：消息默认保存小时：默认为168（24 * 7） - log.retention.minutes：消息默认保存时间，按分钟计 - log.retention.ms：消息设置默认保存时间，按ms计 上述设置中，优先级为`ms`>`minutes`>`hours`，如果优先级较大的被设置，那么取优先级高的设置，在高优先级条目没有被设置时，才取低优先级设置。 > 如果想要设置消息不超时，可以将`log.retention.ms`设置为-1 `log.retention.check.interval.ms`参数可以检查消息是否超时的周期，默认情况下该值设置为5min。 kafka中的日志清除策略由如下两种： - 删除 - 压缩 > 删除 > > 当`log.cleanup.policy`默认值为delete，当该值被设置为delete时，存储超期日志的文件将会被删除。 > > - 基于时间：默认打开。如果segment中所有记录中时间戳最大的记录（最新插入的记录）超过最长时间设置，那么会将该segment删除。（以segment中最新消息时间戳作为segment文件时间戳） > - 基于大小：基于大小的删除默认是关闭的。`log.retention.bytes`默认值为-1，表示对log文件最大的限制，如果单个log文件的大小超过该大小限制，那么会删除log文件对应的最早的segment > 压缩 > > 当`log.cleanup.policy`值设置为compact时，会对超时的segment进行压缩（指segment最新一条插入的消息超时），对于相同key的消息，只会保留最后插入的一条消息，演示如下。 > > | k2 | k1 | k3 | k1 | k2 | > |:-:|:-:|:-:|:-:|:-:| > | 1 | 2 | 3 | 4 | 5 | > > 会被压缩为 > > | k3 | k1 | k2 | > |:-:|:-:|:-:| > | 3 | 4 | 5 | > > 压缩后offset可能并不连续，此时若想要消费的offset不存在，那么会拿到比预期offset大的offset的消息 #### 消费者 通过消息队列的消费方式分为两种： - pull： 拉取 - push： 推送 kafka采用的是拉取模式来进行消费，因为拉取模式可以很好的兼容不同消费者实例的消费速率。各个消费者可以根据自己消费消息的速度来选择拉取消息的速度，如此能避免消息在多个消费者实例间的不合适分配，造成有的消费者实例空闲，而有的消费者实例消息堆积的情况。 ##### kafka消费者工作流程 在kafka架构中，一个分区只能够被同一消费者组中的一个消费者实例进行消费，但是消费者组中的一个消费者实例能够对多个分区进行消费。 并且，任一分区中的消息都会被广播到所有订阅该topic的消费者组中，但是一条消息只能够被消费者组中的一个消费者实例进行消费。 > #### offset > kafka中一个一个分区能够被多个实例进行消费（被多个位于不同消费者组中的消费者进行消费），kafka通过`offset`来记录每个消费者消费到分区的偏移量。 > > offset被存储在kafka的一个主题中(_consumer_offsets)，用于持久化offset数据。 ##### 消费者组 消费者组是由若干个消费者实例组成的集合，位于同一消费者组中的消费者实例，其groupid都相同。 在同一消费者组中，一条消息只能由一个消费者实例进行消费。消费者组中的消费者实例负责消费同一topic中的不同分区，各实例负责的分区没有重叠部分。 > 当消费者组中的消费者实例大于分区数时，消费者组中将存在闲置的消费者实例，同一分区无法由同一消费者组中的多个消费者实例共同消费。 > > ##### 消费者组初始化流程 > 在消费者组中，通过coordinator来协助消费者组实现初始化和分区分配操作，kafka集群中的每个broker实例都有一个对应的coordinator。 > > 消费者组在选择通过哪个coordinator实例来协助进行初始化时，会将groupid根据_consumer_offset主题的分区数量进行取模（默认_consumer_offset主题默认有50个分区，故而默认为groupid%50），根据取模后的结果查找_consumer_offsets的第N号分区位于哪台broker上，该broker的coordinator即会协助消费者组进行初始化。（例如`groupid%50=4`，即会选取_consumer_offsets第四号分区所在的broker来作为协助初始化的broker，该broker上的coordinator会协助消费者组进行初始化。 > > 在消费者组初始化时，组内每个消费者实例都会向选中的coordinator发送消息，请求加入消费者组，而coordinator则会选中其中一个消费者实例作为该消费者组的leader。 > > 选中leader分区之后，coordinator会把broker集群中待消费的topic信息全部发送给leader消费者，leader消费者则是会指定分区分配计划，将topic分区在组内的多个消费者实例之间进行分配。 > > 在leader指定完分区分配计划之后，会将分配计划发送给coordinator，coordinator收到分配方案后则会将分配方案分发给消费者组内的各个消费者实例。 > ##### 消费者组的心跳机制 > 消费者组中的每个消费者都会和coordinator维持长连接，消费者会向coordinator发送心跳包（心跳包默认时间为3s）。 > > 一旦超过指定时间（`session.timeout.ms`，默认为45s）消费者没有向coordinator发送心跳包，那么该消费者将会被从消费者组中移除，并且会触发分区的再平衡，该宕机消费者实例负责的分区将会被分配给同组中别的消费者实例。 > > 如果消费者实例处理消息的时间过长（`max.poll.interval.ms`，默认为5min），自从上一次拉取分区消息时起超过5min还未再次拉取数据，也会触发再平衡。 ##### 消费者实例拉取数据流程 消费者实例在拉取消息时，首先会向broker发送一个fetch请求，从broker处批量拉取数据。其中，fetch过程能够通过如下参数自定义： - `fetch.min.bytes`：单个fetch请求中，broker server返回的最小消息大小。如果消费者向broker发送fetch请求时，可获取的消息没有达到该参数限制的大小，那么broker会等待消息累积，直到累积消息达到`fetch.min.bytes`指定的大小，再将消息批量返回给消费者。`默认情况下，该参数的值为1`，代表fetch请求时消息会被立刻返回。增大该值会批量累积消息提升效率，但是也会增加fetch请求的延迟。 - `fetch.max.bytes`：单个fetch请求中，broker server返回消息最大数据量大小。fetch操作请求数据时，数据将会被broker批量返回。如果如果分区中待返回第一条record的大小大于`fetch.max.bytes`指定的大小，那么该消息大小也会被返回，即使消息大小大于限制。`默认情况下，该参数的默认值为50m`。 - `fetch.max.wait.ms`：在单个请求中，如果broker没有累积到`fetch.max.bytes`指定规模的消息数量，当broker阻塞时间超过`fetch.max.wait.ms`指定的超时时间后，也会将消息返回给消费者。`默认情况下，该参数默认值为500ms`。 在消费者通过fetch请求拉取到数据后，会将拉取的records缓存起来，然后在调用poll方法时返回缓存的record数据。 - `max.poll.records`：该参数用于限制poll()调用返回的最大消息条数。`该参数默认值为500`，调用poll方法时，最多返回500条缓存数据。`max.poll.records`参数用于指定poll行为，并不会对fetch行为造成影响。