ozine's book library

主题

Kafka：从分区日志理解复制、吞吐、Rebalance 与事务 ​

Kafka 的设计适合一类需求：很多事件持续写进来，多套应用分别读取，还要能保留历史、重新处理。理解它时，先把视角放在“每个分区是一条可复制、可重放的日志”上。分区解释并发边界，复制解释故障恢复，消费位置解释重放，事务再把部分读写操作连成一个提交结果。

本文采用 Kafka 4.2 文档和 4.2.0 固定源码作为讲解基线，于 2026-10-13 核对；不声称这是最新主线。尤其注意 4.x 的 KRaft、Consumer Rebalance Protocol、Eligible Leader Replicas 与旧教程的区别。Go 实验运行于 Go 1.27.2、Linux/amd64，仅包含复制条件与事务可见性的简化模型、实际 gzip 压缩实验，没有启动 Kafka，也没有测集群吞吐。

一次读写会碰到哪些角色 ​

名称管什么不要和什么混在一起
Topic / partition一类事件及其中一条有序日志Topic 没有跨分区的统一 offset
分区 Leader接收该分区写入，驱动该分区日志复制不等于整个集群的 active controller
Follower replica从 Leader 拉取同一分区的数据副本用于冗余，不是把业务工作拆成更多份
KRaft controller维护集群元数据、处理分区领导权变化等元数据 Raft quorum 不是每个业务分区都运行一套 Raft
Group coordinator协调消费组与进度等组状态不负责应用数据库的业务事务
Transaction coordinator协调 Kafka 事务状态不能替外部 HTTP 服务回滚

生产者先通过 bootstrap 地址发现元数据，随后把数据发到对应分区 Leader。bootstrap 列表是发现入口，不是一个所有请求都必须经过的代理层。消费者也根据分区和当前元数据请求数据；部署时公布的 Broker 地址必须能被客户端真正访问。[1][2]

正在绘制图示…

可切换适合窗口或原始尺寸
查看图源代码
flowchart TB
    CT["KRaft controller quorum<br/>复制集群元数据"] -. "维护分区 Leader 等元数据" .-> L
    P["Producer"] --> L["Broker A<br/>P0 Leader"]
    L --> F1["Broker B<br/>P0 Follower"]
    L --> F2["Broker C<br/>P0 Follower"]
    L --> G1["billing 组读取 P0"]
    L --> G2["audit 组读取 P0"]
    GC["Group coordinator"] -. "组分配与已提交位置" .-> G1
    GC -. "另一套独立位置" .-> G2
放大图示

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码
flowchart TB
    CT["KRaft controller quorum<br/>复制集群元数据"] -. "维护分区 Leader 等元数据" .-> L
    P["Producer"] --> L["Broker A<br/>P0 Leader"]
    L --> F1["Broker B<br/>P0 Follower"]
    L --> F2["Broker C<br/>P0 Follower"]
    L --> G1["billing 组读取 P0"]
    L --> G2["audit 组读取 P0"]
    GC["Group coordinator"] -. "组分配与已提交位置" .-> G1
    GC -. "另一套独立位置" .-> G2

图中把复制画成 Leader 到 Follower 的数据流；实现上是 Follower 发起 fetch。其他分区可以让 B 或 C 做 Leader，从而把负载分散开。多副本放在同一台宿主机或同一故障域里，不能获得与独立故障域相同的容灾效果。

分区提供并行，也规定有序的范围 ​

分区内的 offset 表示日志位置。生产者把同一业务键映射到同一分区，消费者再按该键的依赖关系处理，才能把分区顺序变成业务顺序。Topic 分成更多分区可以让写入、存储和消费分担到更多节点，但全 Topic 的事件不会因此拥有一个全局顺序。[3]

同一订单应该用稳定的订单 ID 作为 key；所有参与生产的客户端还要使用兼容分区算法。改变分区数时，基于取模的路由可能让某个键进入新分区，原分区里尚未处理完的记录就与新记录并行存在。需要严格连续顺序时，应把扩分区当作路由迁移设计，而不只是一个容量参数。

每个分区的日志又拆成 segment，配合索引定位数据；旧 segment 可以按时间/大小删除。日志压缩（compaction）则保留键的较新状态，适合重建某些状态表，但并非每次写入后立刻只剩一条记录。删除与压缩都不会把 offset 重新编号，所以不能假设相邻业务记录的 offset 差总是 1。[4][5]

副本数、ISR、确认条件分别回答三个问题 ​

副本因子回答“计划保留几份分区数据”。ISR，in-sync replicas，表示被认为仍满足同步要求的一组副本，包含 Leader；它是动态集合。min.insync.replicas 则给确认写入施加最小同步副本数量约束。[6][5:1]

假设 P0 的副本是 A、B、C，A 是 Leader，C 已落后而退出 ISR。目前 ISR 为 A、B：

生产者配置何时返回成功关键风险与代价
acks=0不等待 Broker 确认调用方无法据此知道 Broker 是否收到
acks=1Leader 写入本地日志后即可回复，不等所有同步副本Leader 在复制前故障可能丢失已回复的记录
acks=all等待当前同步副本满足复制确认条件延迟和可用性取决于 ISR 及最小 ISR 等配置

这些配置名和默认值来自官方 Java 客户端配置文档。Go 库可能用不同选项名和默认值，不应直接照搬字符串配置。[7]

若 acks=all、min.insync.replicas=2，A、B 都跟上时可以满足条件；只剩 A 在 ISR 时，就不能把这次写入当作符合要求的成功。反过来，把最小 ISR 设为 1，会允许更弱的同步副本条件，不能再用“三副本”概括每一次成功写入。

下面的函数只做一个静态快照判断：ISR 数量足够，而且每个 ISR 的“下一 offset”都至少是本次写入需要的位置。它故意省略 Leader epoch、ISR 状态切换、High Watermark 推进、ELR 和协议重试，不是 Kafka 协议实现。

go
fmt.Println("ISR A,B caught up:", canAck(2, 101, 101, 101))
fmt.Println("ISR A,B, B behind:", canAck(2, 101, 101, 100))
fmt.Println("only A remains:", canAck(2, 101, 101))
text
ISR A,B caught up: true
ISR A,B, B behind: false
only A remains: false

要求下一位置达到 101，表示至少包含 offset 100。第二行虽然有两份副本，却有一份尚未复制到该位置；第三行只有一份，违反我们设定的数量约束。这个模型帮助检查两个条件，不能证明任何真实 Kafka 故障组合下不会丢数据。

还要分清三个位置：LEO 是某个副本日志的末尾位置，HW（High Watermark）反映复制提交边界，LSO（Last Stable Offset）进一步约束事务读取可见性。它们都不是应用数据库里的业务提交点。acks=all 也不等于每条消息在所有磁盘上分别执行一次同步刷盘；Kafka 的复制与文件系统持久化策略需要一起理解。[6:1][8]

Leader 选举：4.x 不能只背“从 ISR 里选” ​

先看问题本身。若已经承诺的日志前缀到下一位置 101，而某个旧副本只到 98，让它接任会缺失 98、99、100 三个位置：

go
committedNext, staleReplicaNext := 101, 98
fmt.Println("stale leader missing slots:", committedNext-staleReplicaNext)
text
stale leader missing slots: 3

这是示意日志的算术，不是一次实际选举。它说明为什么“先恢复服务，随便挑一个存活副本”可能用数据完整性换可用性。传统 unclean leader election 允许不满足安全同步条件的副本接任，应把数据丢失风险当成明确决策。[6:2]

但在 4.2 上说“ISR 外的副本绝对不能安全成为 Leader”也不准确。Kafka 4.0 引入 Eligible Leader Replicas（ELR），4.1 起新集群默认启用。配合严格的最小 ISR 规则，控制器记录部分虽不在当前 ISR、但仍满足安全接任条件的副本。4.2 文档描述的选择顺序先看 ISR，再看未被 fenced 的 ELR，然后处理最后已知 Leader 的恢复情形。ELR 不是任意落后副本的通行证。[9]

因此，评审故障策略时要核对 Broker 版本、feature level、ELR 是否启用、最小 ISR 和 unclean election 配置。升级后保留旧 feature level 的集群，行为未必等同于同版本全新安装。

KRaft 的选举又是另一层：Controller quorum 用 Raft 维护元数据，通常部署 3 或 5 个 controller，保持多数派可用。Kafka 4.x 已移除 ZooKeeper 运行模式；旧集群迁移应遵循官方路径，不能把 3.x 的 ZooKeeper 运维教程直接用于 4.x。业务分区 Leader 和 controller Leader 是两类角色、两套状态。[10][11]

高吞吐来自少做重复工作 ​

Kafka 并非只靠某一个“零拷贝技巧”就变快。它把很多小消息按分区攒成批次，沿一条较连续的数据路径搬运：

正在绘制图示…

可切换适合窗口或原始尺寸
查看图源代码
flowchart LR
    M["多条记录"] --> B["Producer 按分区组批<br/>可批量压缩"]
    B --> L["Broker 追加日志 segment"]
    L --> P["文件系统页缓存"]
    P --> D["后台写回磁盘"]
    P --> N["读取批次并发送<br/>路径受 TLS 等条件影响"]
    N --> C["Consumer 批量读取、解压"]
放大图示

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码
flowchart LR
    M["多条记录"] --> B["Producer 按分区组批<br/>可批量压缩"]
    B --> L["Broker 追加日志 segment"]
    L --> P["文件系统页缓存"]
    P --> D["后台写回磁盘"]
    P --> N["读取批次并发送<br/>路径受 TLS 等条件影响"]
    N --> C["Consumer 批量读取、解压"]
机制减少了什么需要付出的成本或限制
追加与顺序访问大量随机位置更新、寻址与小块 I/O多分区、压缩清理和追赶旧日志仍会产生其他 I/O
页缓存重复从磁盘读取近期数据，也避免另造一整套对象缓存依赖可用内存；历史回放和工作集超内存会读磁盘
批处理每条记录各发请求、各付固定协议开销攒批时间、内存占用、单批尾延迟
批量压缩重复字段与内容的网络、存储字节压缩/解压 CPU；小而随机的数据未必划算
文件到网络的优化传输明文支持路径上的用户态来回拷贝TLS、转换路径与操作系统条件影响能否使用

这些取舍来自存储和传输结构，不能拿老机械硬盘的数字直接证明今天 NVMe 集群的性能。吞吐还会受复制因子、消息大小、分区分布、压缩算法、网络、消费者和业务处理速度共同限制。[8:1][12]

零拷贝不等于所有数据从不复制 ​

在 4.2.0 固定源码中，FileRecords.writeTo 调用传输层的 transferFrom，明文传输层可利用文件通道的优化路径。设计文档说明，SSL/TLS 路径需要用户态加密，不采用同样的 sendfile 路径。这里的“零拷贝”指减少特定路径上的冗余拷贝，不是网络、内核、压缩、业务解码全部没有拷贝，更不是关闭 TLS 的理由。[13][14][12:1]

把批量压缩的收益单独拿出来试 ​

这段代码把相同的短 JSON 重复 100 次，比较“每条各建一个 gzip 流”与“一起压成一个流”：

go
msg := `{"type":"order.created","status":"new"}` + "\n"
fmt.Println("gzip separately:", 100*compressedSize(msg))
fmt.Println("gzip one batch:", compressedSize(strings.Repeat(msg, 100)))

实际输出，单位为字节：

text
gzip separately: 6500
gzip one batch: 87

单条很短，独立流还要反复付头尾开销；合起来后，重复内容又能被更充分利用。因此这组刻意重复的数据差距很大。它只测 Go 标准库 gzip 输出大小，没有 Kafka record-batch 开销、真实事件差异和网络时间，不能换算成“Kafka 快了 75 倍”。选择 linger、批量大小和压缩算法时，应使用自己的负载测吞吐、P99 延迟和 CPU。[7:1][15]

Rebalance：分区易主时，业务任务不一定已经停下 ​

普通消费组里，每个分区在稳定分配下由组内一个成员负责。增加成员可以分担分区；成员退出、订阅变化和分区变化可能触发重新分配。这个过程叫 Rebalance。[2:1]

假设实例 A 已拿到 P0 的 10、11、12，10 完成并提交下一位置 11，但 11 还在请求数据库。此时 P0 转交 B，B 从 11 读起。如果 A 的旧请求仍然成功，11 就可能由两边都执行。Broker 的所有权变更不会自动撤销已经发往数据库的请求。

情况需要处理的工作不应采取的捷径
分区即将被撤销停止继续派发，按客户端回调语义排空或取消，提交已完成前缀把整个已拉取批次都标成完成
已经失去分区防止旧任务继续提交进度，靠幂等/业务版本处理可能完成的旧副作用假设“提交失败”能回滚已发生业务
一个批次处理太久缩小批次、暂停分区、限制 worker，检查客户端存活机制一味把所有超时调大
频繁重启与抖动查实例资源、心跳、重启策略；评估 static membership给多个实例配置同一个静态成员 ID

并行 worker 要按分区维护实际派发序列的连续完成前缀，不能见到 offset 12 完成就提交 13。失败和取消信号也要回到协调处，避免业务还没结束就认为整个批次安全。

旧资料中的 Rebalance 常指 Classic protocol，其中 eager 分配会撤销大量分区，cooperative sticky 通过增量交接减少迁移和停顿。Kafka 4.0 起 GA 的 Consumer Rebalance Protocol 又进一步把分配逻辑放到服务端，采用全增量设计，不依赖旧式全组同步屏障。4.2 官方 Java 消费者要设置 group.protocol=consumer 才使用新协议；心跳与 session timeout 由相应服务端配置控制。[16]

不要把 Java 的 max.poll.interval.ms、后台心跳行为原样推断到每个 Go SDK。先查库版本是否实现所需 group protocol，以及回调、阻塞 Rebalance、退出方式的具体约定。协议改善停顿，不会替应用消除重复执行窗口。

Exactly-Once：先把“哪一段结果”圈出来 ​

需要分开两项能力：幂等生产者吸收协议重试造成的重复写入；Kafka 事务把一组输出记录和普通消费组的输入进度放进同一次事务提交。幂等生产者不是按业务 payload 去重，应用重新提交一条语义相同的新消息，仍可能产生新记录。[7:2][17]

典型场景是 Kafka → 处理 → Kafka：

正在绘制图示…

可切换适合窗口或原始尺寸
查看图源代码
sequenceDiagram
    participant C as 消费与处理程序
    participant K as Kafka
    participant O as 下游 read_committed 消费者
    C->>K: 读取输入记录
    C->>K: 开始事务，写输出记录
    C->>K: 将输入的下一消费位置纳入同一事务
    alt 成功提交
        K-->>O: 输出事务记录可见
        K-->>C: 输入进度一同提交
    else 事务中止
        K-->>O: 跳过中止事务记录
        K-->>C: 本次输入进度不作为已提交结果前移
    end
放大图示

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码
sequenceDiagram
    participant C as 消费与处理程序
    participant K as Kafka
    participant O as 下游 read_committed 消费者
    C->>K: 读取输入记录
    C->>K: 开始事务，写输出记录
    C->>K: 将输入的下一消费位置纳入同一事务
    alt 成功提交
        K-->>O: 输出事务记录可见
        K-->>C: 输入进度一同提交
    else 事务中止
        K-->>O: 跳过中止事务记录
        K-->>C: 本次输入进度不作为已提交结果前移
    end

事务中止不要求把已经追加的物理字节从日志中抹掉。消费者以 read_committed 读取时过滤中止事务记录，并等待未决事务的可见性边界；read_uncommitted 可以看到随后中止的事务记录。非事务记录按相应普通规则读取。[18]

这也解释了“后面明明写入了，为什么暂时读不到”。模型中 offset 2 的事务未决，即使 offset 3 属于已提交事务，顺序读取也先停在未决边界之前：

go
log := []Record{{0, "committed"}, {1, "aborted"}, {2, "open"}, {3, "committed"}}
fmt.Println("open transaction:", readCommitted(log))
log[2].State = "committed"
fmt.Println("transaction finished:", readCommitted(log))
text
open transaction: [0]
transaction finished: [0 2 3]

offset 1 被跳过，offset 2 提交后，后面的 3 才一起变得可读。这里假定所有记录已达到复制条件，只模拟一个分区的事务可见性；真实事务还包含多分区协调、控制批次、超时和生产者 fencing。

要获得这条链的结果语义，事务生产者需要正确管理稳定且不被并发误用的 transactional.id，禁用与事务方案冲突的消费自动提交，把消费组元数据和下一位置正确纳入事务，并让下游使用 read_committed。Broker 和客户端的事务协议能力也要匹配；Kafka 4.0 的新事务协议已有明确的客户端与服务端启用条件。[17:1][19]

外部数据库更新、发送邮件和支付请求没有自动加入上图。若处理函数先写数据库再提交 Kafka 事务，Kafka 中止不能撤销数据库；重放输入后数据库动作还可能再做。应使用数据库幂等、Outbox/Inbox 或支持协作提交的连接器方案，不能把 Kafka 内部 Exactly-Once 扩大成所有外部副作用恰好一次。

配置评审时，一起看三张清单 ​

第一张是数据边界：哪些键要求有序、保留多久、谁需要重放、哪些外部效果必须幂等。第二张是故障边界：副本放在哪些故障域、ISR 降级时接受停写还是弱化承诺、控制器和磁盘怎样恢复。第三张才是性能参数：分区、批量、压缩、在途上限、消费并发。

若缺少前两张，只给出“3 副本、acks=all、批量 1 MB”仍不足以判断系统是否符合需求。本文没有提供可直接上线的集群配置；端口、认证、磁盘、故障域和已有 feature level 都需要真实环境信息。下面的完整程序只复现本文教学实验。

依据与验证范围 ​

本机运行 go run .、go test -v ./...、go test -race ./...、go vet ./... 均通过。复制测试验证模型不会在副本不足或未跟上时返回成功；可见性测试验证模型跳过中止记录并停在未决记录前。没有进行真实 Kafka 集群、Leader 故障、TLS、Rebalance 或跨分区事务实验；gzip 数字是指定输入的字节数，不是吞吐基准。

完整可复现代码 ​

把以下文件放在同一目录，运行：

sh
go run .
go test -v ./...
go test -race ./...
go vet ./...

go.mod ​

go.mod
module example.com/mq/kafkamodel

go 1.27.2

main.go ​

代码块 · 68 行 · 点击收起 / 展开
go
package main

import (
	"bytes"
	"compress/gzip"
	"fmt"
	"strings"
)

// canAck models a fixed ISR snapshot, not the Kafka replication protocol.
func canAck(minISR, requiredNext int, isrNext ...int) bool {
	if len(isrNext) < minISR {
		return false
	}
	for _, next := range isrNext {
		if next < requiredNext {
			return false
		}
	}
	return true
}

type Record struct {
	Offset int
	State  string
}

func readCommitted(log []Record) []int {
	var result []int
	for _, r := range log {
		if r.State == "open" {
			break
		} // first unstable record blocks later records
		if r.State == "committed" {
			result = append(result, r.Offset)
		}
	}
	return result
}

func compressedSize(s string) int {
	var b bytes.Buffer
	w := gzip.NewWriter(&b)
	if _, err := w.Write([]byte(s)); err != nil {
		panic(err)
	}
	if err := w.Close(); err != nil {
		panic(err)
	}
	return b.Len()
}

func demo() {
	fmt.Println("ISR A,B caught up:", canAck(2, 101, 101, 101))
	fmt.Println("ISR A,B, B behind:", canAck(2, 101, 101, 100))
	fmt.Println("only A remains:", canAck(2, 101, 101))
	committedNext, staleReplicaNext := 101, 98
	fmt.Println("stale leader missing slots:", committedNext-staleReplicaNext)
	log := []Record{{0, "committed"}, {1, "aborted"}, {2, "open"}, {3, "committed"}}
	fmt.Println("open transaction:", readCommitted(log))
	log[2].State = "committed"
	fmt.Println("transaction finished:", readCommitted(log))
	msg := `{"type":"order.created","status":"new"}` + "\n"
	fmt.Println("gzip separately:", 100*compressedSize(msg))
	fmt.Println("gzip one batch:", compressedSize(strings.Repeat(msg, 100)))
}

func main() { demo() }

main_test.go ​

代码块 · 35 行 · 点击收起 / 展开
go
package main

import (
	"reflect"
	"testing"
)

func TestAckNeedsEnoughCaughtUpReplicas(t *testing.T) {
	for _, x := range []struct {
		name     string
		replicas []int
		want     bool
	}{
		{"enough", []int{101, 101}, true},
		{"behind", []int{101, 100}, false},
		{"too-few", []int{101}, false},
	} {
		t.Run(x.name, func(t *testing.T) {
			if got := canAck(2, 101, x.replicas...); got != x.want {
				t.Fatalf("got %v", got)
			}
		})
	}
}

func TestOpenTransactionBlocksLaterCommittedRecord(t *testing.T) {
	log := []Record{{0, "committed"}, {1, "aborted"}, {2, "open"}, {3, "committed"}}
	if got := readCommitted(log); !reflect.DeepEqual(got, []int{0}) {
		t.Fatal(got)
	}
	log[2].State = "committed"
	if got := readCommitted(log); !reflect.DeepEqual(got, []int{0, 2, 3}) {
		t.Fatal(got)
	}
}

  1. Kafka 4.2 Design，The Producer / Load balancing：元数据与直接向分区 Leader 发送。 ↩︎

  2. KafkaConsumer 4.2 API：消费组、位置、Rebalance、多线程处理。 ↩︎ ↩︎

  3. Kafka 4.2 Introduction：Topic、分区、有序范围与副本。 ↩︎

  4. Kafka 4.2 Implementation，Log：segment 与索引。 ↩︎

  5. Kafka 4.2 Topic Configs：最小 ISR、保留与 compaction 配置。 ↩︎ ↩︎

  6. Kafka 4.2 Design，Replication：ISR、复制、故障恢复及 unclean election。ELR 的新增边界另见专门文档。 ↩︎ ↩︎ ↩︎

  7. Kafka 4.2 Producer Configs：acks、幂等、in-flight、批量与压缩。本文不把 Java 默认值归给 Go 客户端。 ↩︎ ↩︎ ↩︎

  8. Kafka 4.2 Design，Persistence：文件系统、页缓存与写入路径。 ↩︎ ↩︎

  9. Kafka 4.2 Eligible Leader Replicas：版本、默认启用范围与选择顺序。 ↩︎

  10. Kafka 4.2 KRaft：角色、controller quorum、动态 quorum 与 feature level。 ↩︎

  11. Kafka 4.2 KRaft vs ZooKeeper：移除和差异。 ↩︎

  12. Kafka 4.2 Design，Efficiency：sendfile、TLS 限制、批量压缩。 ↩︎ ↩︎

  13. Kafka 4.2.0 FileRecords.java，writeTo。 ↩︎

  14. Kafka 4.2.0 PlaintextTransportLayer.java，transferFrom。 ↩︎

  15. Go compress/gzip：本文用本地 Go 1.27.2 实现实际测压缩大小。 ↩︎

  16. Kafka 4.2 Consumer Rebalance Protocol：新旧协议与配置区别。 ↩︎

  17. KafkaProducer 4.2 API：幂等、事务、fencing、sendOffsetsToTransaction。 ↩︎ ↩︎

  18. Kafka 4.2 Consumer Configs，isolation.level：事务可见性和未决事务造成的读取边界。 ↩︎

  19. Kafka 4.2 Transaction Protocol：协议版本及启用条件。 ↩︎

保留各书来源与许可 · 示例运行条件以正文为准