diff --git a/mq/kafka/kafka-尚硅谷.md b/mq/kafka/kafka-尚硅谷.md index 390c0f3..fc2988a 100644 --- a/mq/kafka/kafka-尚硅谷.md +++ b/mq/kafka/kafka-尚硅谷.md @@ -55,4 +55,112 @@ kafka生产者异步发送是指生产者调用kafka客户端接口将发送的 ##### 同步发送 除了异步调用接口外，还可以调用同步调用的接口来发送消息。在调用异步接口发送消息时，仅仅将消息放到缓冲区后调用就返回，可能放到缓冲区的消息并没有发送到broker。可以通过调用同步发送消息的接口来发送消息，其会阻塞并等待broker将消息持久化后返回的异常或者ack应答。 -想要同步发送消息，只需要对异步接口返回的future对象调用`.get`即可，其会等待future对象完成。 \ No newline at end of file +想要同步发送消息，只需要对异步接口返回的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的操作。 + diff --git a/mq/kafka/kafka.md b/mq/kafka/kafka.md index 7732eb9..9827b9b 100644 --- a/mq/kafka/kafka.md +++ b/mq/kafka/kafka.md @@ -346,12 +346,27 @@ follower从leader分区消息消息，就像一个普通的消费者一样，并 另外，**如果一个节点仍然处于活跃状态，但是离同步leader的数据有很长的延迟，那么leader将会将该节点从ISR中移除**。延迟的最长时间通过`replica.lag.time.max.ms`来配置。如果在`replica.lag.time.max.ms`时间内副本没能通过leader到日志结尾的数据，那么副本节点将会从ISR中被移除。 +> `replica.lag.time.max.ms`的默认值为30s + #### 消息提交的定义 在某分区对应ISR中所有的副本都将消息追加到它们的log后，则可以认为该消息被提交。故而，消费者并无需担心分区的leader宕机后消息会丢失的问题，因为消息在提交前已经持久化到其分区副本中。 作为生产者，可以在发送消息时决定是否等待消息被提交，这需要在等待提交所带来的延迟和不等待提交所带来的消息丢失风险中进行权衡。是否等待提交取决于生产者的ack设置。 + +#### acks +可以针对生产者设置acks模式。acks模式可设置为如下值： +- 0：此时生产者发送消息时无需等待broker的ack，消息会被直接添加到缓冲区中，并且消息被认为已发送成功。此时并无法保证消息被发送给broker，并且重试设置也不会起作用 +- 1：这种情况下会等待来自于leader的ack，保证消息被写入到leader分区中。但是，如果leader broker返回ack后立马宕机，其他副本broker并没有同步leader分区数据，那么消息将会被丢失 +- -1（all）：这种情况下，leader broker会等待ISR队列中所有的副本broker都对该消息返回ack后才对生产者返回ack。此时只要当前集群中还存在一个in-sync副本，那么消息就不会丢失 + +> `acks`的默认值是-1（all） + +> acks=-1的情况下会产生重复数据的问题，如果发送消息后，消息已经全部存储到所有的broker，但是再尚未ack的情况下leader宕机，那么生产者会重新发送消息，此时消息会被重复存储在消息队列中 + #### 最小写入副本数 -topic可以设置一个最小写入的副本数，通过配置`min.insync.replicas`，可以对最小写入副本数进行配置。即使消息已经同步到所有ISR副本后，如果同步数目小于该值（同步数目包含leader），消息也无法被视为提交 +当acks设置为-1（all）时，topic可以设置一个最小写入的副本数，通过配置`min.insync.replicas`，可以对最小写入副本数进行配置。即使消息已经同步到所有ISR副本后，如果同步数目小于该值（同步数目包含leader），消息也无法被视为提交 + +> `min.insync.replicas`的值默认为1. 该值只有当生产者ack模式设置为-1（all）时才起作用 > kafka保证一条消息只要被提交，只要有一个in-sync-replica处于活跃状态，那么消息就不会被丢失