diff --git a/notes/消息队列.md b/notes/消息队列.md index 0dc4bdbf..2d38ea69 100644 --- a/notes/消息队列.md +++ b/notes/消息队列.md @@ -19,9 +19,9 @@ * [死信队列](#死信队列) * [七、Kafka](#七kafka) * [分区与消费者组](#分区与消费者组) + * [副本与可靠性](#副本与可靠性) * [存储](#存储) * [为什么吞吐量高](#为什么吞吐量高) - * [副本与可靠性](#副本与可靠性) * [附录：实测](#附录实测) * [A1 批量对吞吐量的影响](#a1-批量对吞吐量的影响) * [A2 重复消费](#a2-重复消费) @@ -234,6 +234,37 @@ Kafka 中的消息按主题（Topic）分类，每个主题分成多个分区（ 消费者加入、退出或者分区数变化时，组内会重新分配分区，称为重平衡（Rebalance）。重平衡期间被重新分配的分区暂停消费，还可能造成重复消费。一个常见的原因是处理一批消息的时间超过了 max.poll.interval.ms（默认 5 分钟），消费者被认为已经失效，它的分区被分给其他消费者。 +### 副本与可靠性 + +每个分区有多个副本，分布在不同的 broker 上，其中一个是 Leader，负责处理读写，其余的 Follower 不断从 Leader 拉取数据。与 Leader 保持同步的副本集合称为 ISR（In-Sync Replicas），Follower 超过 replica.lag.time.max.ms（默认 30 秒）没有追上 Leader 就会被移出 ISR。Kafka 4.0 起完全移除了 ZooKeeper，Leader 选举等元数据由内置的 KRaft 控制器管理，它基于 Raft 协议实现。 + +例如创建 3 个分区、3 个副本的主题，然后用 kill -9 停掉 broker 3（3 个节点的 KRaft 集群实测）： + +```text +停止前 +Partition: 0 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1 +Partition: 1 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2 +Partition: 2 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 +停止后 +Partition: 0 Leader: 2 Replicas: 2,3,1 Isr: 2,1 +Partition: 1 Leader: 1 Replicas: 3,1,2 Isr: 1,2 +Partition: 2 Leader: 1 Replicas: 1,2,3 Isr: 1,2 +``` + +分区 1 的 Leader 从 broker 3 切换到了 ISR 中的 broker 1，其它分区只是 ISR 中少了 3。broker 3 重启并追上数据后，又回到了 ISR 中。 + +生产者的 acks 参数决定写入多少个副本才算成功： + +- acks=0：不等待确认，可能丢消息； +- acks=1：Leader 写入就返回，Leader 宕机而 Follower 还没同步时会丢消息； +- acks=all：ISR 中的副本都写入才返回，Kafka 3.0 起是默认值。 + +acks=all 要配合 min.insync.replicas 使用，否则 ISR 只剩 Leader 一个时，acks=all 就退化成了 acks=1。min.insync.replicas 设为 2 时，ISR 少于 2 个副本，broker 就拒绝 acks=all 的写入。上面停掉 broker 3 之后 ISR 还有 2 个副本，写入正常；把 min.insync.replicas 改为 3 再写入，生产者重试几次后报错 NotEnoughReplicasException，改用 acks=1 则仍然能写入。常用的配置是 3 个副本、min.insync.replicas=2、acks=all，一个副本故障时既不丢消息，也不停止写入。 + +生产者没有收到确认就会重试，这可能让同一条消息写入两次。Kafka 3.0 起默认开启幂等生产者：broker 给每个生产者分配一个 ID，生产者给发往每个分区的消息编上序号，broker 发现序号重复就丢弃。后文查看 .log 文件时，每个批次都带有 producerId: 1001 和 baseSequence 字段。幂等只保证单个分区内不重复；要让写入多个分区的消息同时成功或失败，需要使用事务，消费者设置 isolation.level=read_committed 后只读取已提交事务中的消息。 + +消费者处理完消息后提交偏移量，下次从提交的位置继续读取。默认每 5 秒自动提交一次，已经处理但还没提交时崩溃，重启后就会重复消费。例如消费者读到 3 条扣款消息后被 kill -9，这时组的偏移量仍然为空；它在 session.timeout.ms（默认 45 秒）之后被移出消费者组，重新启动的消费者又收到了这 3 条消息（见[附录 A2](#a2-重复消费)）。因此接收端必须做到幂等，方法见 [接收端的可靠性](#接收端的可靠性)。反过来，如果先提交偏移量再处理，处理失败的消息就丢失了。 + ### 存储 每个分区在磁盘上是一个目录，日志按大小切分成多个段（Segment），文件名是段中第一条消息的偏移量。例如把段的大小设为 1 MB（默认 1 GB），向一个只有一个分区的主题写入 3 万条 100 字节的消息后，目录中的文件和大小如下（Kafka 4.4 实测，省略了 .timeindex 等文件）： @@ -275,37 +306,6 @@ Kafka 中的消息按主题（Topic）分类，每个主题分成多个分区（ 分区并行：一个主题的多个分区分布在不同的 broker 上，读写由多台机器共同承担。 -### 副本与可靠性 - -每个分区有多个副本，分布在不同的 broker 上，其中一个是 Leader，负责处理读写，其余的 Follower 不断从 Leader 拉取数据。与 Leader 保持同步的副本集合称为 ISR（In-Sync Replicas），Follower 超过 replica.lag.time.max.ms（默认 30 秒）没有追上 Leader 就会被移出 ISR。Kafka 4.0 起完全移除了 ZooKeeper，Leader 选举等元数据由内置的 KRaft 控制器管理，它基于 Raft 协议实现。 - -例如创建 3 个分区、3 个副本的主题，然后用 kill -9 停掉 broker 3（3 个节点的 KRaft 集群实测）： - -```text -停止前 -Partition: 0 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1 -Partition: 1 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2 -Partition: 2 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 -停止后 -Partition: 0 Leader: 2 Replicas: 2,3,1 Isr: 2,1 -Partition: 1 Leader: 1 Replicas: 3,1,2 Isr: 1,2 -Partition: 2 Leader: 1 Replicas: 1,2,3 Isr: 1,2 -``` - -分区 1 的 Leader 从 broker 3 切换到了 ISR 中的 broker 1，其它分区只是 ISR 中少了 3。broker 3 重启并追上数据后，又回到了 ISR 中。 - -生产者的 acks 参数决定写入多少个副本才算成功： - -- acks=0：不等待确认，可能丢消息； -- acks=1：Leader 写入就返回，Leader 宕机而 Follower 还没同步时会丢消息； -- acks=all：ISR 中的副本都写入才返回，Kafka 3.0 起是默认值。 - -acks=all 要配合 min.insync.replicas 使用，否则 ISR 只剩 Leader 一个时，acks=all 就退化成了 acks=1。min.insync.replicas 设为 2 时，ISR 少于 2 个副本，broker 就拒绝 acks=all 的写入。上面停掉 broker 3 之后 ISR 还有 2 个副本，写入正常；把 min.insync.replicas 改为 3 再写入，生产者重试几次后报错 NotEnoughReplicasException，改用 acks=1 则仍然能写入。常用的配置是 3 个副本、min.insync.replicas=2、acks=all，一个副本故障时既不丢消息，也不停止写入。 - -生产者没有收到确认就会重试，这可能让同一条消息写入两次。Kafka 3.0 起默认开启幂等生产者：broker 给每个生产者分配一个 ID，生产者给发往每个分区的消息编上序号，broker 发现序号重复就丢弃。前面查看 .log 文件时，每个批次都带有 producerId: 1001 和 baseSequence 字段。幂等只保证单个分区内不重复；要让写入多个分区的消息同时成功或失败，需要使用事务，消费者设置 isolation.level=read_committed 后只读取已提交事务中的消息。 - -消费者处理完消息后提交偏移量，下次从提交的位置继续读取。默认每 5 秒自动提交一次，已经处理但还没提交时崩溃，重启后就会重复消费。例如消费者读到 3 条扣款消息后被 kill -9，这时组的偏移量仍然为空；它在 session.timeout.ms（默认 45 秒）之后被移出消费者组，重新启动的消费者又收到了这 3 条消息（见[附录 A2](#a2-重复消费)）。因此接收端必须做到幂等，方法见 [接收端的可靠性](#接收端的可靠性)。反过来，如果先提交偏移量再处理，处理失败的消息就丢失了。 - ## 附录：实测 环境为 Apple M1 Pro、macOS，Kafka 4.4.0 的 3 个节点的 KRaft 集群运行在同一台机器上（端口 9091～9093，命令中的 $B 为这三个地址）。