Kafka分布式消息系统技术指南
摘要
Apache Kafka是现代分布式架构中的核心消息中间件，以其高吞吐量、低延迟和强一致性著称。本指南从分布式系统基础出发，深入剖析Kafka的技术架构、性能优化原理和生产实践，帮助开发者全面掌握这一关键技术。
内容概览
核心架构篇：分布式存储设计、Topic/Partition模型、副本同步机制、Controller选举原理
消息流转篇：Producer发送流程、Consumer消费机制、Offset管理、批处理优化
可靠性保障篇：ISR机制、故障恢复、事务支持、Exactly-Once语义
性能优化篇：零拷贝技术、压缩算法、网络优化、JVM调优
运维实践篇：集群部署、监控告警、容量规划、故障排查
技术背景
在微服务和云原生架构的推动下，分布式消息系统已成为现代应用架构的重要基石。Kafka凭借其独特的设计理念——基于磁盘的顺序写入、分区并行处理、零拷贝网络传输等技术，实现了传统消息队列难以企及的性能表现。
本指南将带您深入理解这些设计决策背后的技术原理，掌握Kafka在大规模生产环境中的最佳实践。
核心术语对照
为确保阅读的一致性，本文统一使用以下术语：
- Broker（代理节点）：Kafka集群中的服务器节点
- Consumer Group（消费者组）：共同消费Topic的Consumer集合
- Partitioner（分区器）：决定消息发送到哪个分区的组件
- ISR（In-Sync Replicas）：与Leader保持同步的副本集合
- acks确认机制：Producer的消息确认级别配置
- sendfile零拷贝：Linux内核提供的高效文件传输技术
- Offset（偏移量）：消息在分区中的位置标识
第一章：分布式消息系统基础
1.1 消息队列在分布式架构中的作用
在微服务架构中，服务间通信是核心问题。我们以用户注册场景为例分析传统同步RPC调用的问题。
用户注册的业务流程：
- 验证用户信息
- 创建用户账户
- 发送欢迎邮件
- 创建默认设置
- 记录注册统计
传统同步调用方式：
sequenceDiagram
participant Client as 客户端
participant User as 用户服务
participant Email as 邮件服务
participant Setting as 设置服务
participant Stat as 统计服务
Client->>User: 注册请求
User->>User: 1. 验证并创建用户
User->>Email: 2. 同步调用发送邮件
Email-->>User: 邮件发送结果
User->>Setting: 3. 同步调用创建设置
Setting-->>User: 设置创建结果
User->>Stat: 4. 同步调用记录统计
Stat-->>User: 统计记录结果
User->>Client: 注册完成响应
Note over Client,Stat: 串行执行，总耗时 = 各服务耗时之和
为什么是串行的：
- 用户服务必须等待邮件服务返回才能继续
- 必须等待设置服务完成才能调用统计服务
- 任何一个下游服务超时都会阻塞整个流程
- 客户端要等待所有操作完成才能收到响应
同步调用的问题：
- 响应时间长：总响应时间 = 所有下游服务调用时间之和
- 级联故障：邮件服务故障会导致整个注册流程失败
- 紧耦合：用户服务必须知道所有下游服务的接口和地址
- 扩展困难：新增业务逻辑需要修改用户服务代码
1.2 异步消息系统的解决方案
使用消息队列的异步处理方式：
sequenceDiagram
participant Client as 客户端
participant User as 用户服务
participant MQ as 消息队列
participant Email as 邮件服务
participant Setting as 设置服务
participant Stat as 统计服务
Client->>User: 注册请求
User->>User: 1. 验证并创建用户
par 并行发送消息
User->>MQ: 发送"用户注册"消息
and
User->>Client: 2. 立即返回注册成功
end
par 各服务独立消费
MQ->>Email: 消费消息并发送邮件
and
MQ->>Setting: 消费消息并创建设置
and
MQ->>Stat: 消费消息并记录统计
end
Note over Client,Stat: 并行执行，用户无需等待后续处理
为什么是异步的：
- 用户服务创建用户后立即发送消息到队列
- 用户服务不等待下游服务处理完成就返回成功
- 各个下游服务独立从队列消费消息并处理
- 客户端快速收到响应，用户体验更好
异步消息的优势：
- 解耦性：用户服务只需要发消息，不关心谁来处理
- 高可用：某个服务故障不影响用户注册流程
- 弹性伸缩：各服务可以根据负载独立扩展
- 最终一致性：保证消息最终会被处理，但不保证立即处理
需要考虑的问题：
- 数据一致性：如何处理消息处理失败的情况
- 消息顺序：某些业务场景需要保证消息处理顺序
- 重复处理：需要考虑消息重复消费的幂等性问题
1.3 Kafka的技术定位
Apache Kafka是一个分布式流处理平台，具备以下特性：
- 高吞吐量：单机可达数百万TPS
- 低延迟：毫秒级消息传递
- 持久化存储：基于磁盘的可靠存储
- 水平扩展：支持集群动态扩容
第二章：Kafka核心架构
2.1 为什么需要分布式架构
单机消息队列的限制：
- 存储容量限制：单台服务器磁盘容量有限，无法存储海量消息
- 处理能力限制：单CPU无法处理高并发的读写请求
- 可靠性风险：单点故障会导致整个消息系统不可用
- 扩展性差：业务增长时无法灵活扩容
分布式架构的必要性：
graph TB
subgraph "单机限制"
S1[CPU: 有限处理能力]
S2[内存: 有限缓存容量]
S3[磁盘: 有限存储空间]
S4[网络: 单点带宽瓶颈]
end
subgraph "分布式解决方案"
D1[多CPU并行处理]
D2[分布式内存缓存]
D3[水平扩展存储]
D4[负载均衡分担流量]
end
S1 --> D1
S2 --> D2
S3 --> D3
S4 --> D4
2.2 Kafka集群拓扑结构设计原理
在介绍Kafka集群架构之前，需要了解一个重要组件：ZooKeeper。ZooKeeper是一个分布式协调服务，在Kafka集群中承担着关键角色：负责元数据管理、Leader选举、配置管理和集群协调。可以把ZooKeeper理解为Kafka集群的"大脑"，它知道集群中每个组件的状态和位置。
graph TB
subgraph "Kafka Cluster"
B1[Broker 1
负责Partition 0,3]
B2[Broker 2
负责Partition 1,4]
B3[Broker 3
负责Partition 2,5]
end
subgraph "ZooKeeper Ensemble"
Z1[ZK Node 1
Leader选举]
Z2[ZK Node 2
元数据同步]
Z3[ZK Node 3
故障检测]
end
P[Producer
智能路由] --> B1
P --> B2
P --> B3
B1 --> C[Consumer
负载均衡]
B2 --> C
B3 --> C
B1 -.-> Z1
B2 -.-> Z2
B3 -.-> Z3
style B1 fill:#e1f5fe
style B2 fill:#e1f5fe
style B3 fill:#e1f5fe
为什么需要多个Broker：
-
水平扩展存储：
- 每个Broker管理部分Partition
- 总存储容量 = 单Broker容量 × Broker数量
- 可以根据数据量动态增加Broker
-
并行处理提升性能：
- 多个Broker同时处理不同Partition的读写
- 避免单点性能瓶颈
- 理论上处理能力线性增长
-
故障容错：
- Broker故障时，其他Broker继续服务
- 数据副本分布在多个Broker上
- 实现高可用架构
为什么需要ZooKeeper集群：
在分布式系统中，多个节点需要协调工作，这带来了复杂的挑战：
- 节点状态同步问题：
graph TD
B1["Broker 1认为：
Partition A的Leader是Broker 2"]
B2["Broker 2认为：
Partition A的Leader是Broker 3"]
B3["Broker 3认为：
Partition A的Leader是Broker 2"]
style B1 fill:#ffcdd2
style B2 fill:#ffcdd2
style B3 fill:#ffcdd2
- 如果各个Broker对集群状态有不同理解，系统就无法正常工作
- 需要一个权威的协调者来维护统一的状态信息
-
Leader选举的复杂性：
当Partition的Leader Broker故障时：- 谁来决定新的Leader是谁？
- 如何确保所有Broker都知道这个决定？
- 如何防止多个Broker同时认为自己是Leader？
-
配置信息管理：
集群中需要共享的信息包括：- 每个Topic有哪些Partition
- 每个Partition的副本分布在哪些Broker上
- 哪个副本是Leader，哪些是Follower
- Consumer Group的成员信息
为什么ZooKeeper本身也需要集群：
单个ZooKeeper节点会成为整个系统的单点故障：
- 如果ZooKeeper宕机，整个Kafka集群都无法正常工作
- ZooKeeper集群（通常3个或5个节点）提供高可用性
- 即使部分ZooKeeper节点故障，集群仍能正常工作
- ZooKeeper通过一致性算法确保数据的强一致性
为什么Producer连接所有Broker：
-
智能路由：
- Producer获取Topic的Partition分布信息
- 根据分区策略直接连接目标Broker
- 避免消息转发的网络开销
-
负载均衡：
- 不同消息可以并行发送到不同Broker
- 避免单点写入瓶颈
为什么Consumer从所有Broker读取：
-
数据分布：
- Consumer需要的数据可能分布在不同Broker上
- 必须能够访问所有相关Broker
-
故障转移：
- 某个Broker故障时，Consumer可以从其他Broker读取副本数据
2.4 Topic与Partition的层级关系
什么是Topic：
Topic是Kafka中消息的逻辑分类单位。可以理解为一个消息的类别标签。
例如：
user-loginTopic：存放所有用户登录相关的消息order-createdTopic：存放所有订单创建相关的消息payment-completedTopic：存放所有支付完成相关的消息
为什么需要Topic分类：
想象一个系统中有各种类型的消息混合在一起：
graph LR
subgraph "没有Topic分类的混乱状态"
M1[用户登录消息]
M2[订单创建消息]
M3[支付完成消息]
M4[用户登录消息]
M5[库存更新消息]
end
M1 --> M2 --> M3 --> M4 --> M5
C1[订单处理系统] --> M1
note1[需要过滤掉不相关消息]
使用Topic分类后：
graph TB
subgraph "Topic分类管理"
T1[user-login Topic]
T2[order-created Topic]
T3[payment-completed Topic]
end
T2 --> C1[订单处理系统]
note2[直接消费相关消息，无需过滤]
什么是Partition：
Partition是Topic的物理存储分片。一个Topic可以被分成多个Partition。
Topic和Partition的层级关系：
graph TD
T[Topic: user-events
逻辑层面的消息分类]
T --> P0[Partition 0
物理存储分片1]
T --> P1[Partition 1
物理存储分片2]
T --> P2[Partition 2
物理存储分片3]
P0 --> F0[实际文件：/logs/user-events-0/]
P1 --> F1[实际文件：/logs/user-events-1/]
P2 --> F2[实际文件：/logs/user-events-2/]
style T fill:#e3f2fd
style P0 fill:#fff3e0
style P1 fill:#fff3e0
style P2 fill:#fff3e0
层级关系说明：
- Topic是逻辑概念：开发者在代码中指定消息发送到哪个Topic
- Partition是物理概念：消息实际存储在具体的Partition文件中
- 一对多关系：一个Topic包含一个或多个Partition
- Partition编号：从0开始，例如user-events-0、user-events-1、user-events-2
为什么要将Topic分成多个Partition：
- 存储容量突破：
graph TD
subgraph "单Partition限制"
SP[单个Partition] --> SL[单台机器存储限制]
end
subgraph "多Partition扩展"
MP1[Partition 0] --> B1[Broker 1存储]
MP2[Partition 1] --> B2[Broker 2存储]
MP3[Partition 2] --> B3[Broker 3存储]
end
style SL fill:#ffcdd2
style B1 fill:#c8e6c9
style B2 fill:#c8e6c9
style B3 fill:#c8e6c9
- 并行处理能力：
graph LR
subgraph "串行处理（单Partition）"
P[Producer] --> SP[Partition 0]
SP --> C[Consumer]
note1[所有消息串行处理]
end
subgraph "并行处理（多Partition）"
P1[Producer] --> MP1[Partition 0]
P2[Producer] --> MP2[Partition 1]
P3[Producer] --> MP3[Partition 2]
MP1 --> C1[Consumer 1]
MP2 --> C2[Consumer 2]
MP3 --> C3[Consumer 3]
note2[多个消息并行处理]
end
- 负载分散：
- 不同Partition可以分布在不同的Broker上
- 避免单个Broker成为瓶颈
- 实现真正的分布式存储
消息如何分配到Partition：
当Producer发送消息时，需要决定消息发送到哪个Partition：
graph TD
M[消息] --> D{分配策略}
D -->|有Key| H[根据Key的hash值分配
相同Key的消息进入同一Partition]
D -->|无Key| R[轮询分配
消息均匀分布到各Partition]
H --> P1[Partition 0]
R --> P1
R --> P2[Partition 1]
R --> P3[Partition 2]
这样的层级设计让Kafka既有逻辑上的清晰分类，又有物理上的分布式存储能力。
Partition的副本机制设计：
graph TD
T[Topic: user-events] --> P1[Partition 0]
T --> P2[Partition 1]
T --> P3[Partition 2]
P1 --> R1[Leader Replica
Broker 1]
P1 --> R2[Follower Replica
Broker 2]
P1 --> R3[Follower Replica
Broker 3]
P2 --> R4[Leader Replica
Broker 2]
P2 --> R5[Follower Replica
Broker 1]
P2 --> R6[Follower Replica
Broker 3]
P3 --> R7[Leader Replica
Broker 3]
P3 --> R8[Follower Replica
Broker 1]
P3 --> R9[Follower Replica
Broker 2]
style R1 fill:#ff9999
style R4 fill:#ff9999
style R7 fill:#ff9999
为什么需要副本机制：
-
数据持久性保证：
- 磁盘故障是常见硬件问题
- 单副本存在数据丢失风险
- 多副本提供数据冗余保护
-
高可用性要求：
- Broker故障时服务不能中断
- 副本分布在不同Broker上
- Leader故障时Follower可以接替
-
读写分离的可能性：
- Leader处理写请求，保证数据一致性
- Follower可以处理读请求（新版本支持）
- 分担Leader的读取压力
为什么要区分Leader和Follower：
-
一致性保证：
- 如果所有副本都可以写入，会产生数据冲突
- Leader统一处理写入，确保数据顺序一致
- Follower从Leader同步，保证副本一致性
-
性能优化：
- 避免多点写入的分布式锁开销
- Leader可以批量同步给多个Follower
- 简化了副本间的协调复杂度
副本分布策略的考虑：
- 跨Broker分布：避免单点故障
- 跨机架分布：避免机架级故障
- 负载均衡：Leader副本均匀分布在各Broker
2.5 消息的物理存储模型
graph TB
subgraph "Partition Directory"
subgraph "Segment 1"
S1L[00000000000.log]
S1I[00000000000.index]
S1T[00000000000.timeindex]
end
subgraph "Segment 2"
S2L[00000100000.log]
S2I[00000100000.index]
S2T[00000100000.timeindex]
end
subgraph "Active Segment"
S3L[00000200000.log]
S3I[00000200000.index]
S3T[00000200000.timeindex]
end
end
Log Segment：
- 每个Partition被分割为多个Segment文件
- 只有最新的Segment可写，其他为只读
- 便于数据清理和管理
索引机制：
- Offset Index：消息偏移量到文件位置的映射
- Time Index：时间戳到消息偏移量的映射
- 支持快速随机访问和时间范围查询
第三章：消息生产与消费机制
3.1 Producer消息发送流程
sequenceDiagram
participant P as Producer
participant I as Interceptor
participant S as Serializer
participant PT as Partitioner
participant A as Accumulator
participant ST as Sender Thread
participant B as Broker
P->>I: 1. 消息预处理
I->>S: 2. 序列化Key/Value
S->>PT: 3. 计算目标分区
PT->>A: 4. 加入消息批次
A->>ST: 5. 批次发送
ST->>B: 6. 网络传输
B->>ST: 7. 响应确认
ST->>P: 8. 回调处理
关键组件的具体作用：
Interceptor（消息拦截器）：
在消息发送前进行预处理的组件。
具体例子：
1 | 原始消息：{"userId": "123", "action": "login"} |
Serializer（序列化器）：
将Java对象转换为字节数组，以便网络传输。
具体例子：
1 | Java对象：User user = new User("张三", 25); |
Partitioner（分区器）：
决定消息发送到哪个具体分区。
具体例子：
1 | 场景：用户操作消息需要保证同一用户的消息有序 |
RecordAccumulator（消息累加器）：
将多个消息打包成批次，提高发送效率。
具体例子：
1 | 时间线： |
Sender Thread（独立发送线程）：
专门负责网络I/O的后台线程。
具体例子：
1 | 主线程的工作： |
3.2 分区器(Partitioner)策略
graph TD
M[Message] --> PK{"Has Key?"}
PK -->|Yes| H["Hash(key) % partitions"]
PK -->|No| RR["Round Robin"]
H --> P1[Partition 0]
RR --> P2[Partition 1]
RR --> P3[Partition 2]
分区策略：
- Hash分区：基于消息Key的哈希值分区，确保相同Key的消息有序
- 轮询分区：无Key消息的均匀分布策略
- 自定义分区：实现Partitioner接口的业务逻辑
3.3 Consumer Group机制的设计原理
为什么需要Consumer Group：
在实际系统中，我们经常遇到这样的需求：
场景1：水平扩展消费能力
1 | 订单Topic每秒产生1000条消息 |
如果没有Consumer Group机制：
graph TD
T[订单Topic
1000条/秒] --> C1[Consumer A
100条/秒]
T --> C2[Consumer B
100条/秒]
T --> C3[Consumer C
100条/秒]
note1[问题：每个Consumer都会收到所有1000条消息
造成重复处理]
style C1 fill:#ffcdd2
style C2 fill:#ffcdd2
style C3 fill:#ffcdd2
使用Consumer Group后：
graph TD
subgraph "订单处理组（Group: order-processors）"
C1[Consumer A
处理333条/秒]
C2[Consumer B
处理333条/秒]
C3[Consumer C
处理334条/秒]
end
T[订单Topic
1000条/秒] --> P1[Partition 0]
T --> P2[Partition 1]
T --> P3[Partition 2]
P1 --> C1
P2 --> C2
P3 --> C3
note2[每条消息只被处理一次
总处理能力：1000条/秒]
style C1 fill:#c8e6c9
style C2 fill:#c8e6c9
style C3 fill:#c8e6c9
场景2：多业务系统共享数据
1 | 用户注册消息需要被多个系统处理： |
Consumer Group的核心规则：
-
组内不重复：同一Group内的Consumer不会重复消费同一条消息
1
2
3
4
5订单处理组内：
- Consumer A 处理 订单1、订单4、订单7...
- Consumer B 处理 订单2、订单5、订单8...
- Consumer C 处理 订单3、订单6、订单9...
每个订单只被处理一次 -
组间独立：不同Group可以独立消费所有消息
1
2
3
4
5用户注册Topic的消息：
- 邮件处理组：所有注册消息都会被处理
- 统计分析组：所有注册消息都会被处理
- 推荐系统组：所有注册消息都会被处理
每个Group都能看到完整的数据
Consumer与Partition的分配机制：
graph TD
subgraph "Topic: user-events (3个Partition)"
P1[Partition 0]
P2[Partition 1]
P3[Partition 2]
end
subgraph "Group A: 邮件处理组 (3个Consumer)"
CA1[Consumer A1]
CA2[Consumer A2]
CA3[Consumer A3]
end
subgraph "Group B: 统计分析组 (2个Consumer)"
CB1[Consumer B1]
CB2[Consumer B2]
end
P1 --> CA1
P2 --> CA2
P3 --> CA3
P1 --> CB1
P2 --> CB1
P3 --> CB2
note1[Group A：一对一分配]
note2[Group B：Consumer B1处理2个分区]
分配规则的具体逻辑：
-
Consumer数量 ≤ Partition数量：
1
2
33个Partition，2个Consumer：
- Consumer 1：负责Partition 0 + Partition 1
- Consumer 2：负责Partition 2 -
Consumer数量 > Partition数量：
1
2
3
4
5
63个Partition，5个Consumer：
- Consumer 1：负责Partition 0
- Consumer 2：负责Partition 1
- Consumer 3：负责Partition 2
- Consumer 4：空闲（无分区分配）
- Consumer 5：空闲（无分区分配）
为什么要这样设计：
可能有人会问：为什么不让多个Consumer同时处理同一个分区来提高并行度？答案是为了保证消息的顺序性。如果多个Consumer同时处理同一分区，就无法保证消息按照发送顺序被处理，这在很多业务场景下是不可接受的。
重要澄清：分区之间没有顺序关系
这里需要澄清一个关键概念：分区之间是没有顺序关系的，只有分区内部的消息是有序的。Consumer可以并行地从多个分区读取数据，不需要按照分区编号顺序去一个一个读取。
graph TD
subgraph "Topic: user-events"
P0["Partition 0
消息A1→A2→A3"]
P1["Partition 1
消息B1→B2→B3"]
P2["Partition 2
消息C1→C2→C3"]
end
subgraph "Consumer Group"
C1["Consumer 1"]
C2["Consumer 2"]
C3["Consumer 3"]
end
P0 --> C1
P1 --> C2
P2 --> C3
note1[分区内有序：消息按顺序处理, 例如 A1→A2→A3]
note2[分区间无序: A1, B1, C1 的处理顺序是随机的
并行消费: 三个 Consumer 同时工作]
P1 -.-> note1
C2 -.-> note2
-
保证消费顺序：
- 分区内有序：单个Partition内的消息严格按照写入顺序消费
- 分区间无序：不同Partition之间的消息没有顺序关系
- 同一Partition只能被一个Consumer消费：确保分区内的顺序性
- 并行处理：多个Consumer可以同时从不同分区消费数据
具体例子说明顺序性：
假设有一个订单处理场景：
1 | 时间轴上的事件： |
关键点：
- 需要保序的消息（同一用户的操作）必须发送到同一分区
- 不同用户的消息可以发送到不同分区，并行处理
- Consumer不需要等待其他分区，可以各自独立处理
深入理解：分区内容的差异性
您提出了一个关键问题：分区之间的内容是一样的还是不一样的？答案是：分区之间的内容是完全不同的，它们存储的是不同的消息。
graph TD
subgraph "Producer发送消息"
M1[消息1: 用户A下单]
M2[消息2: 用户B下单]
M3[消息3: 用户A付款]
M4[消息4: 用户C下单]
end
subgraph "分区器决策"
PD[根据用户ID哈希分配]
end
subgraph "Topic分区存储"
P0[Partition 0
用户A下单
用户A付款]
P1[Partition 1
用户B下单
用户C下单]
end
M1 --> PD
M2 --> PD
M3 --> PD
M4 --> PD
PD --> P0
PD --> P1
style P0 fill:#e3f2fd
style P1 fill:#fff3e0
分区设计的业务考量：
这里有一个重要的设计原则：分区间的顺序确实必须不影响业务逻辑。这是为什么呢？
-
水平扩展的前提：
1
2
3
4如果业务依赖跨分区的顺序，那么：
- 无法并行处理（必须串行）
- 无法水平扩展（增加分区没意义）
- 性能无法提升（回到单线程模式） -
业务设计要求：
1
2
3
4良好的分区策略应该确保：
- 相关的消息在同一分区（如同一用户的操作）
- 无关的消息在不同分区（如不同用户的操作）
- 业务逻辑不依赖跨分区的消息顺序
正确的分区策略示例：
✅ 好的设计：电商订单系统
1 | 分区策略：按用户ID哈希分区 |
❌ 错误的设计：库存管理系统
1 | 如果设计成： |
分区与副本的区别：
还需要澄清一个常见的混淆：分区(Partition)和副本(Replica)是不同的概念：
graph TD
subgraph "Topic: orders"
P0[Partition 0
用户A的消息]
P1[Partition 1
用户B的消息]
end
subgraph "Broker 1"
P0L[Partition 0 Leader
用户A的消息]
P1F1[Partition 1 Follower
用户B的消息副本]
end
subgraph "Broker 2"
P0F1[Partition 0 Follower
用户A的消息副本]
P1L[Partition 1 Leader
用户B的消息]
end
P0 --> P0L
P0 --> P0F1
P1 --> P1L
P1 --> P1F1
style P0L fill:#4caf50
style P1L fill:#4caf50
- 分区：存储不同的消息内容，用于并行处理
- 副本：存储相同的消息内容，用于容错备份
重要澄清：Broker与分区的关系
您提出了一个关键问题：一个分区可能非常大，一台机器放不下怎么办？这里需要澄清一个重要概念：
一个分区只能存储在一台机器上，不能跨机器分割。但是，Kafka通过以下机制来解决存储限制问题：
graph TD
subgraph "Topic: user-events（3个分区）"
P0[Partition 0]
P1[Partition 1]
P2[Partition 2]
end
subgraph "Kafka集群"
subgraph "Broker 1（机器1）"
P0_B1[Partition 0
存储在此机器]
P1_R1[Partition 1 副本]
end
subgraph "Broker 2（机器2）"
P1_B2[Partition 1
存储在此机器]
P2_R1[Partition 2 副本]
end
subgraph "Broker 3（机器3）"
P2_B3[Partition 2
存储在此机器]
P0_R1[Partition 0 副本]
end
end
P0 --> P0_B1
P0 --> P0_R1
P1 --> P1_B2
P1 --> P1_R1
P2 --> P2_B3
P2 --> P2_R1
style P0_B1 fill:#4caf50
style P1_B2 fill:#4caf50
style P2_B3 fill:#4caf50
核心原则：
- 每个分区主副本只存在于一个Broker上
- 不同分区可以分布在不同Broker上
- 通过增加分区数量来水平扩展存储
如何处理分区过大问题：
当单个分区变得过大时，Kafka提供了几种解决方案：
-
数据保留策略：
1
2
3
4自动清理机制：
- 基于时间：保留7天内的数据，自动删除过期数据
- 基于大小：分区超过100GB时，删除最老的数据
- 基于压缩：对相同Key的消息进行压缩，只保留最新值 -
Log Segment分片：
1
2
3
4
5
6
7
8
9
10单个分区在磁盘上分为多个Segment文件：
- 每个Segment默认1GB大小
- 只有最新的Segment可写
- 旧的Segment可以独立删除或归档
例如：
/kafka-logs/topic-0/
├── 00000000000000000000.log (1GB, 已满)
├── 00000000000001000000.log (1GB, 已满)
├── 00000000000002000000.log (800MB, 当前写入) -
分区数量规划：
1
2
3
4
5
6
7
8设计时就要考虑分区数量：
错误示例：
- 1个分区存储100TB数据 ❌
正确示例：
- 100个分区，每个1TB数据 ✅
- 分布在50台机器上，每台2个分区
Broker与分区的具体关系：
graph LR
subgraph B1 ["Broker 1 (192.168.1.10)"]
B1P1["orders-0/ (主分区)"]
B1P2["orders-2/ (副本)"]
B1P3["users-1/ (副本)"]
end
subgraph B2 ["Broker 2 (192.168.1.11)"]
B2P1["orders-1/ (主分区)"]
B2P2["orders-0/ (副本)"]
B2P3["users-0/ (主分区)"]
end
subgraph B3 ["Broker 3 (192.168.1.12)"]
B3P1["orders-2/ (主分区)"]
B3P2["orders-1/ (副本)"]
B3P3["users-1/ (主分区)"]
end
B1P1 -.->|副本| B2P2
B2P1 -.->|副本| B3P2
B3P1 -.->|副本| B1P2
B2P3 -.->|副本| B1P3
B3P3 -.->|副本| B1P3
style B1P1 fill:#4caf50
style B2P1 fill:#4caf50
style B3P1 fill:#4caf50
style B2P3 fill:#4caf50
style B3P3 fill:#4caf50
关键理解：
-
分区不能跨Broker分割：
- 每个分区的完整数据只能存在一个Broker上
- 不能把一个分区的一部分放在Broker1，另一部分放在Broker2
-
Broker可以托管多个分区：
- 一个Broker可以存储多个不同分区的主副本
- 一个Broker也可以存储多个其他分区的从副本
-
水平扩展的正确方式：
- 不是让单个分区跨多台机器
- 而是创建更多分区，分布到更多机器上
实际容量规划示例：
假设您有一个日志收集系统，预计存储需求：
1 | 业务需求： |
分区大小的经验法则：
1 | 推荐的单分区大小： |
-
简化offset管理：
- 每个Consumer只需要管理自己负责的Partition的offset
- 避免复杂的消息分发和同步机制
-
故障自动恢复：
- Consumer故障时，其负责的Partition会自动分配给其他Consumer
- 新Consumer加入时，会自动重新分配Partition
- 实现了自动的负载均衡和故障转移
3.4 消息定位与Offset概念
为什么需要消息定位机制：
在分布式消息系统中，Consumer需要知道：
- 我已经读到了哪里？
- 下次应该从哪里开始读？
- 如果我重启了，应该从哪里继续？
Offset的本质：
Offset是消息在Partition中的位置标识，类似于数组的索引：
graph TD
M0["消息0
offset=0"] --> M1["消息1
offset=1"] --> M2["消息2
offset=2"] --> M3["消息3
offset=3"] --> M4["消息4
offset=4"]
style M2 fill:#ffeb3b
note1["Consumer当前位置：offset=2
下次读取：offset=3"]
Offset的关键特性：
- 单调递增：新消息的offset总是比旧消息大
- 分区独立：每个Partition有独立的offset序列
- 永久有效：offset在消息存在期间始终有效
3.5 Consumer的消费位置追踪
Consumer如何知道读取位置：
sequenceDiagram
participant C as Consumer
participant B as Broker
participant OS as Offset Storage
Note over C,OS: Consumer启动时
C->>OS: 1. 查询上次提交的offset
OS->>C: 2. 返回offset=100
Note over C,OS: 正常消费流程
C->>B: 3. 从offset=100开始拉取
B->>C: 4. 返回消息[100,101,102]
C->>C: 5. 处理业务逻辑
C->>OS: 6. 提交新offset=103
OS->>C: 7. 确认提交成功
Note over C,OS: Consumer重启后
C->>OS: 8. 再次查询offset
OS->>C: 9. 返回offset=103
C->>B: 10. 从offset=103继续拉取
为什么需要Offset提交：
- 状态持久化：Consumer重启后能够继续上次的进度
- 避免重复处理：已处理的消息不会再次处理
- 集群容错：Consumer故障时其他实例可以接替
3.6 Offset提交策略的选择
自动提交的工作原理：
sequenceDiagram
participant C as Consumer
participant B as Broker
participant T as Timer
loop 每5秒自动提交
C->>B: 拉取消息
B->>C: 返回消息批次
C->>C: 开始处理消息
T->>C: 定时器触发(5秒到)
C->>B: 自动提交当前offset
Note over C,B: 无论消息是否处理完成
end
自动提交的问题场景：
场景1：消息丢失
sequenceDiagram
participant C as Consumer
participant B as Broker
C->>B: 拉取消息offset=100-102
B->>C: 返回3条消息
Note over C: 定时器到期，自动提交offset=103
C->>B: 提交offset=103
Note over C: Consumer崩溃，消息未处理
Note over C: 重启后从offset=103开始
Note over C: 消息100-102永远丢失
场景2：重复消费
sequenceDiagram
participant C as Consumer
participant B as Broker
C->>B: 拉取消息offset=100-102
B->>C: 返回3条消息
C->>C: 处理完所有消息
Note over C: Consumer崩溃，offset未提交
Note over C: 重启后仍从offset=100开始
C->>B: 重新拉取offset=100-102
Note over C: 消息100-102被重复处理
手动提交的精确控制：
sequenceDiagram
participant C as Consumer
participant B as Broker
participant BL as Business Logic
C->>B: 拉取消息offset=100
B->>C: 返回消息
C->>BL: 处理业务逻辑
BL->>C: 处理成功
C->>B: 手动提交offset=101
B->>C: 提交确认
Note over C,B: 只有业务处理成功才提交offset
两种策略的适用场景：
| 提交方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 自动提交 | 简单易用，无需编码 | 可能丢失或重复消息 | 对消息精确性要求不高的场景 |
| 手动提交 | 精确控制，避免丢失 | 编码复杂，需要异常处理 | 对消息精确性要求高的场景 |
Offset存储位置的演进：
- 早期版本：存储在ZooKeeper中，但ZK不适合高频写入
- 现在版本：存储在Kafka内部Topic
__consumer_offsets中- 利用Kafka自身的持久化和副本机制
- 减少对ZooKeeper的依赖
- 提供更好的性能和可靠性
第四章：可靠性与一致性保证
4.1 Kafka的基础存储原理：Append-Only Log
在理解Kafka的副本机制之前，我们需要先理解Kafka最核心的设计理念：追加式日志（Append-Only Log）。这是Kafka高性能和可靠性的基石。
什么是Append-Only Log：
Append-Only Log是一种只能在末尾添加数据，不能修改已有数据的存储结构：
graph LR
subgraph "传统数据库的更新操作"
DB1[记录1: 用户A, 余额100] --> DB2[记录1: 用户A, 余额150]
DB3[记录2: 用户B, 余额200] --> DB4[记录2: 用户B, 余额180]
note1[直接修改原有记录]
end
subgraph "Kafka的Append-Only Log"
L1[消息1: 用户A登录] --> L2[消息2: 用户A购买商品]
L2 --> L3[消息3: 用户B登录]
L3 --> L4[消息4: 用户A退款]
note2[只能追加，不能修改]
end
style DB2 fill:#fff3e0
style DB4 fill:#fff3e0
style L4 fill:#e3f2fd
为什么选择Append-Only设计：
-
磁盘顺序写入的性能优势：
现代磁盘的性能特点是：顺序写入比随机写入快几十倍甚至上百倍。
1
2
3
4典型磁盘性能数据：
- 顺序写入：100-200 MB/s
- 随机写入：1-5 MB/s
- 性能差距：20-200倍Kafka利用这个特点，所有消息都追加写入到日志末尾，实现了极高的写入性能。
-
简化并发控制：
为什么传统数据库不使用append-only设计？因为传统数据库需要支持随机更新和删除操作（如修改用户信息、删除过期数据），而消息队列的特点是数据一旦发送就不需要修改，这使得append-only设计成为可能且高效：
graph TD
subgraph "传统数据库的并发问题"
T1[线程1: 修改记录A] --> Lock1[需要加锁]
T2[线程2: 修改记录A] --> Lock2[等待锁释放]
Lock1 --> Conflict[可能产生冲突]
end
subgraph "Append-Only的并发优势"
T3[线程1: 追加消息] --> A1[写入位置1000]
T4[线程2: 追加消息] --> A2[写入位置1001]
A1 --> Simple[无需加锁，天然有序]
A2 --> Simple
end
style Conflict fill:#ffcdd2
style Simple fill:#c8e6c9
- 数据不可变性带来的优势：
- 故障恢复简单：数据一旦写入就不会改变，不存在部分更新的问题
- 备份一致性：副本只需要复制新增的数据，不用担心数据被修改
- 缓存友好：操作系统可以安全地缓存数据，不用担心缓存失效
Kafka中Log的物理结构：
每个Partition在磁盘上就是一个append-only的日志文件：
graph TD
subgraph "Partition的物理存储"
subgraph "Log Segment 1 (已满)"
S1[00000000000000000000.log
offset 0-999]
end
subgraph "Log Segment 2 (已满)"
S2[00000000000001000000.log
offset 1000-1999]
end
subgraph "Active Segment (当前写入)"
S3[00000000000002000000.log
offset 2000-当前]
Arrow[新消息追加到这里 →]
end
end
S1 --> S2 --> S3
style S3 fill:#e3f2fd
Log Segment的设计原理：
为什么要将一个Partition分成多个Segment？
-
文件大小管理：
- 单个文件过大会影响操作系统的文件操作性能
- 默认1GB一个Segment，便于管理和备份
-
数据清理效率：
graph LR
Old[过期的旧Segment] --> Delete[直接删除整个文件]
Active[活跃Segment] --> Keep[继续保留]
note1[删除整个文件比删除文件中的部分内容要快得多]
- 索引文件配套：
每个Segment都有对应的索引文件，便于快速定位消息：1
2
300000000000001000000.log - 消息数据
00000000000001000000.index - offset索引
00000000000001000000.timeindex - 时间索引
消息在Log中的存储格式：
graph LR
subgraph "单条消息的结构"
Offset[8字节
Offset]
Size[4字节
消息大小]
CRC[4字节
校验码]
Magic[1字节
版本号]
Attr[1字节
属性]
Key[变长
消息Key]
Value[变长
消息内容]
end
Append-Only Log的读取机制：
既然数据只能追加，Kafka如何实现高效的读取？
sequenceDiagram
participant C as Consumer
participant I as Index File
participant L as Log File
C->>I: 1. 查找offset=1500的消息
I->>I: 2. 二分查找最接近的索引项
I->>C: 3. 返回：offset=1500在文件位置45000
C->>L: 4. 从位置45000开始读取
L->>C: 5. 返回消息内容
索引文件记录了部分offset与文件位置的映射关系（稀疏索引），Consumer可以快速定位到目标消息的大概位置，然后顺序读取。
4.2 基于Append-Only Log的副本同步机制
理解了Kafka的基础存储原理后，我们就能更好地理解副本同步是如何工作的。
副本同步的本质：
由于Kafka使用append-only log，副本同步就是将Leader的日志文件复制到Follower上的过程：
sequenceDiagram
participant P as Producer
participant L as Leader
participant F1 as Follower 1
participant F2 as Follower 2
P->>L: 1. 发送消息
L->>L: 2. 追加到本地Log(offset=100)
par Follower主动拉取
F1->>L: 3. 拉取请求"我当前offset=99"
L->>F1: 4. 返回offset=100的消息
F1->>F1: 5. 追加到本地Log(offset=100)
and
F2->>L: 3. 拉取请求"我当前offset=99"
L->>F2: 4. 返回offset=100的消息
F2->>F2: 5. 追加到本地Log(offset=100)
end
par 确认同步完成
F1->>L: 6. ACK: 已同步到offset=100
F2->>L: 6. ACK: 已同步到offset=100
end
L->>P: 7. 确认消息已写入所有副本
副本同步的关键特点：
-
Follower主动拉取：
- Follower会定期向Leader发送拉取请求
- 请求中包含自己当前的offset位置
- Leader返回这个offset之后的所有新消息
-
顺序一致性：
- 由于append-only的特性，所有副本的消息顺序完全一致
- Follower按照Leader的顺序追加消息
- 不存在消息顺序不一致的问题
-
增量同步：
- Follower只需要拉取自己缺失的消息
- 不需要重新传输已有的消息
- 提高了同步效率
ISR (In-Sync Replicas) 机制：
ISR是Kafka用来标识与Leader保持同步状态的副本集合，这是Kafka保证数据一致性的核心机制。在分布式环境中，网络延迟、节点性能差异等因素会导致某些Follower无法及时跟上Leader的更新进度。ISR机制就是用来区分哪些副本是"可靠同步"的，哪些是"延迟过大"的：
graph TD
subgraph "Partition的所有副本"
L[Leader
offset=1000]
F1[Follower 1
offset=1000]
F2[Follower 2
offset=998]
F3[Follower 3
offset=850]
end
subgraph "ISR集合"
L2[Leader
延迟=0]
F1_2[Follower 1
延迟=0]
F2_2[Follower 2
延迟=2条消息]
end
subgraph "OSR (Out-of-Sync)"
F3_2[Follower 3
延迟=150条消息]
end
L --> L2
F1 --> F1_2
F2 --> F2_2
F3 --> F3_2
style L2 fill:#c8e6c9
style F1_2 fill:#c8e6c9
style F2_2 fill:#fff3e0
style F3_2 fill:#ffcdd2
ISR的动态维护：
ISR不是固定的，会根据Follower的同步情况动态调整。为了理解这个过程，我们需要先明确一个概念：
OSR (Out-of-Sync Replicas)：
OSR是指那些因为各种原因（网络延迟、处理能力不足、节点故障等）无法与Leader保持同步的Follower副本。
graph TD
subgraph "所有副本状态"
AR[AR: All Replicas
所有副本]
AR --> ISR[ISR: In-Sync Replicas
同步副本]
AR --> OSR[OSR: Out-of-Sync Replicas
非同步副本]
end
ISR --> List1[Leader + 同步的Follower]
OSR --> List2[延迟过大的Follower]
style ISR fill:#c8e6c9
style OSR fill:#ffcdd2
动态调整的判断标准：
1 | 主要参数：replica.lag.time.max.ms（默认30秒） |
动态调整过程：
- 正常状态：Follower定期拉取消息，保持在ISR中
- 出现延迟：Follower因故障或网络问题延迟拉取
- 超过阈值：延迟时间超过
replica.lag.time.max.ms→ 从ISR中移除 → 成为OSR - 恢复同步：OSR中的Follower恢复正常，追上Leader进度 → 重新加入ISR
这种动态机制确保ISR中始终是真正能够可靠同步的副本，为数据一致性提供保障。
4.3 生产者确认机制的设计原理
在分布式系统中，Producer发送消息到Broker后面临一个关键问题：如何确认消息真正被安全存储了？
问题的复杂性：
考虑这样的场景：Producer发送了一条重要的订单消息到Kafka，然后继续处理其他业务。但是：
- 消息可能在网络传输中丢失
- Broker可能在接收消息时发生故障
- 消息可能写入了Leader但还没同步到Follower，Leader就宕机了
如果Producer不知道消息是否真正安全存储，就无法保证数据的可靠性。
Kafka的解决方案：acks确认机制
Kafka通过acks参数让Producer可以选择不同的确认级别，在性能和可靠性之间做权衡：
graph TD
P[Producer发送消息] --> Choice{选择确认级别}
Choice -->|acks=0| Fast[追求最高性能]
Choice -->|acks=1| Balance[平衡性能和可靠性]
Choice -->|acks=all| Safe[追求最高可靠性]
Fast --> Risk1[风险：可能丢失消息]
Balance --> Risk2[风险：Leader故障时可能丢失]
Safe --> Risk3[风险：性能相对较低]
style Fast fill:#ffeb3b
style Balance fill:#ff9800
style Safe fill:#4caf50
acks=0：追求极致性能的选择
sequenceDiagram
participant P as Producer
participant N as Network
participant B as Broker
P->>N: 发送消息
P->>P: 立即认为发送成功
Note over P: 不等待任何确认
N->>B: 消息可能到达Broker
Note over N,B: 也可能在网络中丢失
为什么选择acks=0：
- 极致性能需求：日志收集、监控指标等场景，需要极高的吞吐量
- 可容忍丢失：偶尔丢失几条日志或监控数据不会影响整体分析
- 成本考虑：网络和存储资源有限，愿意用少量数据丢失换取性能
acks=1：平衡性能与可靠性
sequenceDiagram
participant P as Producer
participant L as Leader
participant F1 as Follower 1
participant F2 as Follower 2
P->>L: 发送消息
L->>L: 写入本地日志
L->>P: 确认消息已写入
Note over P: Producer认为发送成功
par 后台同步过程
L->>F1: 同步消息（异步）
L->>F2: 同步消息（异步）
end
acks=1的风险场景：
1 | 时间线： |
为什么选择acks=1：
- 常见的平衡选择：大多数业务场景的默认选择
- 性能可接受：只需等待Leader确认，延迟较低
- 可靠性足够：Leader故障概率相对较低
acks=all/-1：追求数据安全的选择
sequenceDiagram
participant P as Producer
participant L as Leader
participant F1 as Follower 1
participant F2 as Follower 2
P->>L: 发送消息
L->>L: 写入本地日志
par ISR同步过程
L->>F1: 同步消息
F1->>F1: 写入本地日志
F1->>L: 确认同步完成
and
L->>F2: 同步消息
F2->>F2: 写入本地日志
F2->>L: 确认同步完成
end
L->>P: 所有ISR副本都确认后才回复
Note over P: Producer才认为发送成功
acks=all的安全保障：
即使Leader宕机，消息也已经在ISR的其他副本中安全存储，新选出的Leader一定包含这条消息。
acks=all的潜在问题：
1 | 问题场景：如果ISR中只有Leader一个副本 |
三种级别的适用场景总结：
| acks级别 | 性能 | 可靠性 | 适用场景 | 典型应用 |
|---|---|---|---|---|
| 0 | 最高 | 最低 | 可容忍数据丢失 | 日志收集、指标监控 |
| 1 | 中等 | 中等 | 一般业务场景 | 用户行为追踪、非关键业务 |
| all | 最低 | 最高 | 关键数据不能丢失 | 订单处理、支付记录、审计日志 |
理解了这三种确认级别的设计原理和适用场景，开发者就能根据具体业务需求选择合适的可靠性策略，在性能和数据安全之间找到最佳平衡点。
4.3 故障恢复机制
graph TD
F[Leader故障] --> D[检测故障]
D --> C[Controller选举新Leader]
C --> U[更新元数据]
U --> N[通知所有Broker]
N --> S[服务恢复]
style F fill:#ff6666
style S fill:#66ff66
Controller选举：
- 基于ZooKeeper的分布式锁机制
- Controller负责管理分区状态和副本分配
- 故障时自动选举新的Controller
ZooKeeper分布式锁的工作原理：
在Kafka集群中，只能有一个Broker担任Controller角色。如何在多个Broker中选出唯一的Controller？这就需要依赖ZooKeeper的分布式锁机制。
sequenceDiagram
participant B1 as Broker 1
participant B2 as Broker 2
participant B3 as Broker 3
participant ZK as ZooKeeper
Note over B1,ZK: 集群启动，各Broker竞争Controller
par 并发竞争Controller
B1->>ZK: 尝试创建 /controller 节点
B2->>ZK: 尝试创建 /controller 节点
B3->>ZK: 尝试创建 /controller 节点
end
ZK->>B1: 创建成功，成为Controller
ZK->>B2: 创建失败，节点已存在
ZK->>B3: 创建失败，节点已存在
Note over B1: Broker 1 成为Controller
Note over B2,B3: Broker 2,3 监听Controller变化
B2->>ZK: 监听 /controller 节点删除事件
B3->>ZK: 监听 /controller 节点删除事件
ZooKeeper分布式锁的核心特性：
-
原子性操作：
1
2
3
4ZooKeeper的create操作是原子的：
- 要么创建成功，要么失败
- 不存在部分创建的中间状态
- 确保只有一个客户端能成功创建相同路径的节点 -
临时节点机制：
为什么Controller必须使用临时节点而不能使用永久节点？如果使用永久节点，当Controller故障时节点不会自动删除，其他Broker就无法感知到Controller的故障，整个集群会陷入无主状态。临时节点与ZooKeeper会话绑定，一旦会话中断就会自动删除，这样其他Broker立即感知到故障并开始新的选举：
graph TD
Controller[Controller Broker] --> Session[ZK Session]
Session --> Node["/controller 临时节点"]
Session --> Heartbeat[定期发送心跳]
Heartbeat --> Alive{会话保活}
Alive -->|正常| Keep[保持节点存在]
Alive -->|超时/故障| Delete[自动删除节点]
Delete --> Trigger[触发选举]
style Node fill:#e3f2fd
style Delete fill:#ffcdd2
style Trigger fill:#fff3e0
- 会话超时检测：
1
2
3
4
5ZooKeeper会话机制：
- Controller与ZooKeeper维持TCP长连接
- 定期发送心跳包（默认1/3 session timeout）
- 如果心跳超时，ZooKeeper认为Controller故障
- 自动删除临时节点，其他Broker收到通知开始新一轮选举
Controller选举的详细流程：
flowchart TD
Start[Broker启动] --> Check{检查/controller节点}
Check -->|不存在| Create[尝试创建临时节点]
Check -->|存在| Watch[监听节点变化]
Create --> Success{创建成功?}
Success -->|是| BeController[成为Controller]
Success -->|否| Watch
BeController --> LoadMeta[加载集群元数据]
LoadMeta --> StartLeaderElection[启动Leader选举]
StartLeaderElection --> ManagePartitions[管理分区状态]
Watch --> NodeDeleted{节点被删除?}
NodeDeleted -->|是| Create
NodeDeleted -->|否| Continue[继续监听]
Continue --> NodeDeleted
style BeController fill:#4caf50
style Watch fill:#ff9800
style NodeDeleted fill:#ffeb3b
Controller故障恢复过程：
当Controller Broker发生故障时：
sequenceDiagram
participant Controller as Controller Broker
participant ZK as ZooKeeper
participant B2 as Broker 2
participant B3 as Broker 3
Note over Controller: Controller故障/网络分区
Controller->>ZK: 心跳中断(连接断开)
Note over ZK: session timeout到期
ZK->>ZK: 删除 /controller 临时节点
ZK->>B2: 通知:/controller 节点删除
ZK->>B3: 通知:/controller 节点删除
par 新一轮Controller竞争
B2->>ZK: 尝试创建 /controller
B3->>ZK: 尝试创建 /controller
end
ZK->>B2: 创建失败
ZK->>B3: 创建成功,成为新Controller
B3->>B3: 加载集群元数据
B3->>B3: 重新进行Leader选举
B3->>B2: 通知新的分区Leader信息
Note over B2,B3: 集群恢复正常运行
分布式锁机制的优势：
-
避免脑裂问题：
1
2
3
4
5什么是脑裂：多个节点同时认为自己是主节点
ZooKeeper解决方案：
- 临时节点只能有一个
- 其他节点通过监听机制感知主节点状态
- 主节点故障时自动触发重新选举 -
自动故障检测：
1
2
3
4
5
6
7
8
9传统心跳机制的问题：
- 需要额外的心跳检测线程
- 心跳间隔和超时时间难以调优
- 网络波动可能导致误判
ZooKeeper会话机制的优势：
- 基于TCP连接的天然心跳
- 会话超时由ZooKeeper统一管理
- 更可靠的故障检测机制 -
事件驱动的通知机制：
graph LR
Event[Controller节点删除] --> Watcher1[Broker 2 收到通知]
Event --> Watcher2[Broker 3 收到通知]
Event --> Watcher3[Broker N 收到通知]
Watcher1 --> Action1[立即参与选举]
Watcher2 --> Action2[立即参与选举]
Watcher3 --> Action3[立即参与选举]
这种基于ZooKeeper的分布式锁机制确保了Kafka集群中Controller角色的唯一性和高可用性，是Kafka分布式协调的核心基础。
第五章：性能优化原理
Kafka作为高吞吐量的消息系统，在性能优化方面运用了多种底层技术。理解这些优化技术的原理，对于合理配置和调优Kafka集群至关重要。
5.1 I/O性能瓶颈与零拷贝技术
高并发场景下的I/O挑战：
在高吞吐量的消息系统中，Kafka需要频繁地从磁盘读取消息并通过网络发送给Consumer。这个过程看似简单，但在操作系统层面却涉及复杂的数据拷贝流程。
传统I/O操作的性能问题：
让我们分析一下Consumer从Kafka读取消息的传统流程：
sequenceDiagram
participant C as Consumer
participant K as Kafka Broker
participant OS as 操作系统
participant Disk as 磁盘
C->>K: 请求消息数据
K->>OS: read()系统调用
OS->>Disk: 从磁盘读取数据
Disk->>OS: 数据读入内核缓冲区
OS->>K: 数据拷贝到用户空间
K->>OS: write()系统调用
OS->>OS: 数据拷贝到Socket缓冲区
OS->>C: 通过网络发送数据
这个过程中，数据经历了4次拷贝和2次上下文切换：
graph TD
subgraph "传统I/O的数据拷贝路径"
D[磁盘数据] --> K1[内核缓冲区
第1次拷贝]
K1 --> U[用户空间缓冲区
第2次拷贝]
U --> K2[Socket内核缓冲区
第3次拷贝]
K2 --> N[网络接口卡
第4次拷贝]
end
subgraph "上下文切换开销"
Context1[内核态 → 用户态]
Context2[用户态 → 内核态]
end
style U fill:#ffcdd2
style Context1 fill:#ffcdd2
style Context2 fill:#ffcdd2
性能问题分析：
-
CPU资源浪费：
1
2
3
4每次数据拷贝都需要CPU参与：
- 从内核缓冲区拷贝到用户空间：CPU密集操作
- 从用户空间拷贝到Socket缓冲区：再次占用CPU
- 大量消息传输时，CPU大部分时间用于数据拷贝而非业务处理 -
内存带宽消耗：
1
2
3
4内存总线带宽有限：
- 多次拷贝占用宝贵的内存带宽
- 影响其他程序的内存访问性能
- 在高并发场景下成为系统瓶颈 -
上下文切换开销：
1
2
3
4用户态和内核态切换代价：
- 保存/恢复CPU寄存器状态
- 清空CPU缓存（TLB失效）
- 每次切换消耗数百个CPU周期
零拷贝技术的解决方案：
为了解决这些问题，Linux内核提供了零拷贝技术。Kafka使用的是sendfile系统调用：
sequenceDiagram
participant C as Consumer
participant K as Kafka Broker
participant OS as 操作系统
participant Disk as 磁盘
C->>K: 请求消息数据
K->>OS: sendfile()系统调用
Note over OS: 一次系统调用完成整个传输
OS->>Disk: 读取数据到内核缓冲区
OS->>OS: 直接从内核缓冲区发送到网络
OS->>C: 通过网络发送数据
零拷贝的优化效果：
graph TD
subgraph "零拷贝的数据路径"
D2[磁盘数据] --> K3[内核缓冲区
第1次拷贝]
K3 --> N2[网络接口卡
第2次拷贝]
end
subgraph "性能提升"
Benefit1[减少2次CPU拷贝]
Benefit2[减少2次上下文切换]
Benefit3[节省内存带宽]
Benefit4[降低CPU使用率]
end
style D2 fill:#c8e6c9
style N2 fill:#c8e6c9
style Benefit1 fill:#e8f5e8
style Benefit2 fill:#e8f5e8
style Benefit3 fill:#e8f5e8
style Benefit4 fill:#e8f5e8
sendfile系统调用的工作原理：
Kafka的零拷贝技术主要通过Linux内核提供的sendfile系统调用实现。sendfile是专门为高效文件传输设计的系统调用，它允许数据直接在内核空间中从文件传输到网络Socket，完全绕过用户空间：
1 | // 传统方式需要多次系统调用 |
Kafka中零拷贝的应用场景：
-
Consumer拉取消息：
- Consumer请求历史消息时
- Broker直接从Log文件发送数据到Consumer
- 无需将消息内容加载到JVM堆内存
-
副本同步：
- Follower从Leader同步消息时
- Leader直接从Log文件发送到Follower
- 提高副本同步效率
零拷贝技术的限制：
需要注意的是，零拷贝并非在所有场景都适用：
1 | 不适用零拷贝的情况： |
性能提升的量化效果：
在实际测试中，零拷贝技术能够带来显著的性能提升：
1 | 测试场景：发送100MB数据 |
5.2 网络传输效率与批处理优化
高频网络调用的性能瓶颈：
在消息系统中，如果每条消息都单独发送，会面临严重的性能问题。让我们分析一个具体场景：
问题场景：
1 | 电商系统在促销期间： |
单条消息发送的开销：
sequenceDiagram
participant P as Producer
participant N as Network
participant B as Broker
loop 每条消息都要重复这个过程
P->>N: TCP包头 + 消息1 (约100字节数据 + 40字节开销)
N->>B: 网络传输
B->>N: ACK响应 (40字节开销)
N->>P: 确认收到
Note over P,B: 单条消息实际传输140字节，开销占57%
end
网络开销分析：
每次网络调用的固定开销包括：
1 | 1. TCP/IP协议开销： |
批处理技术的解决思路：
批处理的核心思想是：将多条消息打包成一个批次发送，减少网络调用次数。
graph TD
subgraph "单条发送模式"
M1[消息1] --> S1[网络调用1]
M2[消息2] --> S2[网络调用2]
M3[消息3] --> S3[网络调用3]
M4[消息4] --> S4[网络调用4]
S1 --> Cost1[开销: 54字节]
S2 --> Cost2[开销: 54字节]
S3 --> Cost3[开销: 54字节]
S4 --> Cost4[开销: 54字节]
end
subgraph "批处理模式"
M5[消息1] --> Batch[批次缓冲区]
M6[消息2] --> Batch
M7[消息3] --> Batch
M8[消息4] --> Batch
Batch --> S5[单次网络调用]
S5 --> Cost5[开销: 54字节]
end
style Cost5 fill:#c8e6c9
style Cost1 fill:#ffcdd2
style Cost2 fill:#ffcdd2
style Cost3 fill:#ffcdd2
style Cost4 fill:#ffcdd2
Kafka Producer的批处理实现：
flowchart TD
Start[消息到达] --> CheckBatch{检查当前批次}
CheckBatch -->|批次不存在| CreateBatch[创建新批次]
CheckBatch -->|批次存在| AddToBatch[添加到批次]
CreateBatch --> AddToBatch
AddToBatch --> CheckTrigger{检查触发条件}
CheckTrigger -->|batch.size已满| SendNow[立即发送]
CheckTrigger -->|linger.ms超时| SendTimer[定时发送]
CheckTrigger -->|条件未满足| WaitMore[等待更多消息]
SendNow --> NetworkCall[网络传输]
SendTimer --> NetworkCall
WaitMore --> Start
NetworkCall --> Response[等待响应]
style SendNow fill:#4caf50
style SendTimer fill:#ff9800
style WaitMore fill:#2196f3
关键批处理参数详解：
-
batch.size（批次大小阈值）：
1
2
3
4
5
6
7
8
9
10
11
12作用：控制每个批次的最大字节数
默认值：16384字节（16KB）
工作原理：
- 当批次中消息的总大小达到batch.size时，立即发送
- 即使linger.ms时间未到，也会触发发送
- 适合高吞吐量场景，确保网络带宽充分利用
调优建议：
- 高吞吐量：增大到32KB或64KB
- 低延迟要求：减小到4KB或8KB
- 网络带宽有限：适当减小避免网络拥塞 -
linger.ms（等待时间阈值）：
1
2
3
4
5
6
7
8
9
10
11
12作用：控制批次等待时间的最大值
默认值：0毫秒（立即发送）
工作原理：
- 即使批次未满，等待linger.ms时间后也会发送
- 给更多消息机会加入批次，提高批处理效率
- 在延迟和吞吐量之间做权衡
调优建议：
- 高吞吐量优先：设置为5-100ms
- 低延迟要求：保持默认值0ms
- 均衡场景：设置为10-20ms
批处理的性能提升效果：
让我们通过具体数据来理解批处理的价值：
graph LR
subgraph "批处理前（单条发送）"
Scenario1[10,000条消息/秒]
Calls1[10,000次网络调用]
Overhead1[540KB协议开销]
Latency1[10秒网络延迟累积]
end
subgraph "批处理后（100条/批次）"
Scenario2[10,000条消息/秒]
Calls2[100次网络调用]
Overhead2[5.4KB协议开销]
Latency2[0.1秒网络延迟]
end
style Calls2 fill:#c8e6c9
style Overhead2 fill:#c8e6c9
style Latency2 fill:#c8e6c9
实际测试数据对比：
1 | 测试环境：单Producer，发送1MB消息到Kafka |
批处理策略的权衡考虑：
graph TD
BatchSize[批处理配置] --> Throughput{吞吐量优化}
BatchSize --> Latency{延迟优化}
BatchSize --> Memory{内存使用}
Throughput -->|增大batch.size| HighThroughput[更高吞吐量
但延迟增加]
Throughput -->|增大linger.ms| MoreBatching[更好的批处理效果
但延迟增加]
Latency -->|减小batch.size| LowLatency[更低延迟
但吞吐量下降]
Latency -->|减小linger.ms| QuickSend[更快发送
但批处理效果差]
Memory -->|批次过大| MemoryPressure[内存压力增大]
Memory -->|批次过小| CPUOverhead[CPU开销增大]
style HighThroughput fill:#4caf50
style LowLatency fill:#ff9800
style MemoryPressure fill:#f44336
5.3 存储空间优化与压缩技术
大规模消息存储的挑战：
在实际生产环境中，Kafka需要存储海量的消息数据。让我们分析一个具体场景来理解存储压力：
典型场景分析：
1 | 大型互联网公司的日志收集系统： |
存储成本问题：
graph TD
DataVolume[8.64TB/天] --> StorageCost{存储成本}
StorageCost --> SSD[SSD存储
$0.1/GB/月]
StorageCost --> HDD[机械硬盘
$0.03/GB/月]
SSD --> SSDCost[月成本：$25,920]
HDD --> HDDCost[月成本：$7,776]
DataVolume --> NetworkCost{网络传输成本}
NetworkCost --> Bandwidth[带宽消耗
100MB/秒]
Bandwidth --> BandwidthCost[专线成本：$5000/月]
style SSDCost fill:#ffcdd2
style HDDCost fill:#fff3e0
style BandwidthCost fill:#ffcdd2
压缩技术的必要性：
压缩技术可以显著减少存储和网络传输成本：
1 | 假设平均压缩比为3:1 |
不同压缩算法的特性对比：
为什么Kafka需要支持多种压缩算法？这是因为不同的应用场景有不同的优化目标：有些场景更关注存储成本（如日志归档），有些更关注实时性（如在线交易），有些需要在两者间平衡（如用户行为分析）。因此，在选择压缩算法时，需要根据具体需求在压缩比、CPU消耗、延迟之间做权衡：
graph TD
subgraph "压缩算法特性对比"
GZIP[gzip
压缩比：高
CPU：高
延迟：高]
SNAPPY[snappy
压缩比：中
CPU：中
延迟：中]
LZ4[lz4
压缩比：中低
CPU：低
延迟：低]
ZSTD[zstd
压缩比：最高
CPU：可调
延迟：可调]
end
subgraph "应用场景"
HighCompression[存储成本敏感
CPU资源充足]
Balanced[均衡场景
中等压缩需求]
LowLatency[延迟敏感
实时处理]
Flexible[灵活配置
最佳压缩]
end
GZIP --> HighCompression
SNAPPY --> Balanced
LZ4 --> LowLatency
ZSTD --> Flexible
style GZIP fill:#ff9800
style SNAPPY fill:#4caf50
style LZ4 fill:#2196f3
style ZSTD fill:#9c27b0
详细压缩算法分析：
-
GZIP压缩算法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14技术原理：
- 基于DEFLATE算法，结合LZ77和Huffman编码
- 通过查找重复字符串模式实现压缩
- 适合文本数据和结构化数据
性能特点：
- 压缩比：通常达到3-5:1，最高可达10:1
- CPU消耗：高，压缩和解压都比较耗时
- 内存使用：中等，需要维护字典
适用场景：
- 离线数据处理和存储
- 存储成本是主要考虑因素
- CPU资源充足，延迟要求不严格 -
Snappy压缩算法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14技术原理：
- 由Google开发，专注于压缩/解压速度
- 使用简化的LZ77算法
- 牺牲一定压缩比换取性能
性能特点：
- 压缩比：通常2-3:1
- CPU消耗：中等，平衡了速度和压缩比
- 内存使用：低，算法简单
适用场景：
- 实时数据处理
- CPU和延迟都有一定要求
- Kafka的默认推荐选择 -
LZ4压缩算法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14技术原理：
- 极速压缩算法，专注于速度
- 简化的字典查找机制
- 解压速度极快
性能特点：
- 压缩比：1.5-2.5:1
- CPU消耗：很低，适合高并发场景
- 内存使用：很低
适用场景：
- 超低延迟要求
- 高并发实时处理
- CPU资源受限环境 -
ZSTD压缩算法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14技术原理：
- 由Facebook开发的现代压缩算法
- 可调节的压缩级别（1-22）
- 结合了多种优化技术
性能特点：
- 压缩比：2-8:1（根据压缩级别）
- CPU消耗：可调节，从低到高
- 内存使用：可配置
适用场景：
- 需要灵活配置压缩策略
- 追求最佳压缩比
- 新版本Kafka的推荐选择
Kafka中压缩机制的工作流程：
sequenceDiagram
participant P as Producer
participant B as Broker
participant C as Consumer
Note over P: Producer端压缩
P->>P: 1. 收集消息到批次
P->>P: 2. 对整个批次进行压缩
P->>B: 3. 发送压缩后的数据
Note over B: Broker端存储
B->>B: 4. 直接存储压缩数据
Note over B: 不解压，保持压缩状态
Note over C: Consumer端解压
C->>B: 5. 拉取压缩数据
B->>C: 6. 返回压缩数据
C->>C: 7. 解压获得原始消息
压缩策略的配置建议：
根据不同业务场景选择合适的压缩算法：
flowchart TD
Start[选择压缩算法] --> Priority{主要优化目标}
Priority -->|存储成本| Storage[存储优化]
Priority -->|网络带宽| Network[带宽优化]
Priority -->|处理延迟| Latency[延迟优化]
Priority -->|CPU资源| CPU[CPU优化]
Storage --> StorageChoice[推荐：GZIP或ZSTD
高压缩比，节省存储]
Network --> NetworkChoice[推荐：GZIP或ZSTD
减少网络传输]
Latency --> LatencyChoice[推荐：LZ4或Snappy
快速压缩解压]
CPU --> CPUChoice[推荐：LZ4
最低CPU消耗]
style StorageChoice fill:#4caf50
style NetworkChoice fill:#4caf50
style LatencyChoice fill:#2196f3
style CPUChoice fill:#ff9800
压缩效果的实测数据：
以JSON格式的日志数据为例：
1 | 测试数据：10MB JSON日志文件 |
压缩技术的最佳实践：
-
Producer端配置：
1
2
3
4
5
6
7
8# 启用压缩
compression.type=snappy
# 配合批处理使用
batch.size=32768
linger.ms=10
# 原因：批次越大，压缩效果越好 -
Topic级别配置：
1
2
3
4
5
6# 在Topic创建时指定压缩类型
compression.type=gzip
# 清理策略配合压缩
cleanup.policy=delete
retention.bytes=10737418240 # 10GB -
监控指标：
1
2
3
4
5
6关键指标：
- 压缩前后数据大小比率
- Producer压缩延迟
- Consumer解压延迟
- CPU使用率变化
- 存储空间节省量
通过合理选择和配置压缩算法，可以在保证性能的前提下显著降低Kafka集群的存储和网络成本。
第六章：集群管理与运维
6.1 分区重平衡机制
sequenceDiagram
participant C1 as Consumer 1
participant C2 as Consumer 2
participant C3 as New Consumer
participant GC as Group Coordinator
C3->>GC: Join Group
GC->>C1: Rebalance Signal
GC->>C2: Rebalance Signal
C1->>GC: Leave Group
C2->>GC: Leave Group
GC->>GC: Partition Assignment
GC->>C1: New Assignment
GC->>C2: New Assignment
GC->>C3: New Assignment
重平衡触发条件：
- Consumer加入或离开Consumer Group
- Topic的Partition数量变化
- Consumer超时未发送心跳
6.2 监控指标体系
graph TD
subgraph "Broker指标"
B1[消息吞吐量]
B2[请求延迟]
B3[磁盘使用率]
B4[网络使用率]
end
subgraph "Topic指标"
T1[消息生产速率]
T2[消息消费速率]
T3[消费延迟]
T4[分区分布]
end
subgraph "Consumer指标"
C1[消费Lag]
C2[重平衡频率]
C3[处理耗时]
C4[错误率]
end
关键性能指标：
- Producer Metrics：发送速率、错误率、批次大小
- Broker Metrics：磁盘I/O、网络I/O、副本同步延迟
- Consumer Metrics：消费lag、重平衡时间、处理延迟
6.3 容量规划
graph TD
R[业务需求] --> T[吞吐量计算]
R --> S[存储容量计算]
R --> N[网络带宽计算]
T --> P[分区数规划]
S --> D[磁盘规划]
N --> B[Broker数量规划]
P --> Config[集群配置]
D --> Config
B --> Config
规划要素：
- 分区数量：基于并发消费需求和单分区性能限制
- 副本因子：基于可用性要求和存储成本
- 存储配置：基于消息保留时间和磁盘I/O性能
第七章：故障排查与问题解决
7.1 消息丢失排查
flowchart TD
Start[消息丢失] --> PC{Producer配置检查}
PC -->|acks!=all| P1[设置acks=all]
PC -->|正常| BC{Broker配置检查}
BC -->|副本不足| B1[增加副本因子]
BC -->|正常| CC{Consumer配置检查}
CC -->|自动提交| C1[改为手动提交]
CC -->|正常| LC[检查日志文件]
排查步骤：
- 检查Producer的acks和retries配置
- 验证Broker的副本同步状态
- 确认Consumer的offset提交策略
- 分析Broker和Consumer的错误日志
7.2 性能瓶颈诊断
graph TD
PB[性能瓶颈] --> CPU{CPU使用率}
PB --> MEM{内存使用率}
PB --> DISK{磁盘I/O}
PB --> NET{网络I/O}
CPU -->|高| C1[优化序列化
减少GC]
MEM -->|高| M1[调整JVM参数
优化缓存]
DISK -->|高| D1[增加分区
优化存储]
NET -->|高| N1[启用压缩
增加带宽]
性能调优策略：
- JVM调优：合理设置堆大小和GC参数
- 操作系统调优：调整文件描述符限制和网络参数
- Kafka配置调优：优化批处理、压缩、副本同步参数
7.3 数据一致性问题
sequenceDiagram
participant P as Producer
participant L as Leader
participant F as Follower
participant C as Consumer
P->>L: Send Message (offset=100)
L->>L: Write to Log
L->>F: Replicate
Note over L,F: 网络分区发生
L->>C: High Watermark=99
Note over L,F: Follower选为新Leader
F->>C: High Watermark=98
Note over C: 数据回滚
一致性保证机制：
- High Watermark：确保只消费已被所有ISR确认的消息
- Leader Epoch：防止日志截断导致的数据不一致
- 幂等性Producer：避免重复消息导致的数据异常
第八章：高级特性与扩展
8.1 事务支持
sequenceDiagram
participant P as Producer
participant TC as Transaction Coordinator
participant B1 as Broker 1
participant B2 as Broker 2
P->>TC: Begin Transaction
TC->>P: Transaction ID
P->>B1: Send Message (TXN)
P->>B2: Send Message (TXN)
P->>TC: Commit Transaction
TC->>B1: Commit Marker
TC->>B2: Commit Marker
事务特性：
- 原子性：多分区写入的原子性保证
- 隔离性：读取已提交的事务消息
- 幂等性：避免重复发送导致的数据重复
8.2 Exactly-Once语义
graph TD
EOS[Exactly-Once Semantics] --> IP[幂等性Producer]
EOS --> T[事务支持]
EOS --> RI[消费端幂等性]
IP --> PID[Producer ID]
IP --> SN[Sequence Number]
T --> TID[Transaction ID]
T --> TCO[Transaction Coordinator]
RI --> UO[业务唯一标识]
RI --> DT[外部存储去重]
实现机制：
- Producer端：PID + Sequence Number实现幂等性
- 跨分区：事务机制保证原子性
- Consumer端：业务层面的去重逻辑
8.3 流处理集成
graph LR
K1[Kafka Topic A] --> KS[Kafka Streams]
K2[Kafka Topic B] --> KS
KS --> K3[Kafka Topic C]
KS --> K4[Kafka Topic D]
subgraph "Stream Processing"
KS --> F[Filter]
KS --> M[Map]
KS --> A[Aggregate]
KS --> J[Join]
end
Kafka Streams特性：
- 无需额外集群，直接基于Kafka Client
- 支持窗口操作、状态存储、故障恢复
- 与Kafka生态深度集成
总结与实践建议
核心技术要点
- 架构理解：分布式存储、副本机制、分区策略
- 性能优化：零拷贝、批处理、压缩算法
- 可靠性保证：ISR机制、故障恢复、事务支持
- 运维监控：指标体系、容量规划、故障排查
技术栈集成建议
graph TB
subgraph "数据采集层"
A1[应用日志] --> Kafka
A2[业务事件] --> Kafka
A3[监控指标] --> Kafka
end
subgraph "流处理层"
Kafka --> KS[Kafka Streams]
Kafka --> Flink[Apache Flink]
Kafka --> Storm[Apache Storm]
end
subgraph "存储层"
KS --> ES[Elasticsearch]
Flink --> HBase[HBase]
Storm --> Redis[Redis]
end
持续学习路径
- 深入源码：理解核心算法和数据结构
- 性能调优：JVM调优、操作系统调优、网络优化
- 生态工具：Kafka Connect、Schema Registry、Confluent Platform
- 流处理框架：Kafka Streams、Apache Flink、Apache Storm
通过系统性学习Kafka的技术原理和实践应用，能够在分布式系统设计和微服务架构中发挥其最大价值。
All articles on this blog are licensed under CC BY-NC-SA 4.0 unless otherwise stated.
Related Articles
2024-10-09
如何成为优秀的软件工程师
全面的软件工程师成长指南：深入讲解数据结构与算法的实际应用，详述设计模式在服务器开发中的运用，介绍良好的编程习惯与高效工具链，帮助开发者从基础技能到架构设计全方位提升编程能力。
2024-11-18
传输和备份实践
详细介绍了rsync、SCP等文件传输工具的使用方法和实际应用场景，包括本地同步、远程传输、加密备份等最佳实践，帮助开发者选择合适的数据传输和备份解决方案。
2024-11-18
安全加密实践-GPG
全面介绍GPG加密工具的安装配置、密钥管理、文件加密解密、数字签名等核心功能，涵盖跨平台使用指南和对称加密实践，为开发者提供完整的数据安全保护方案。
2024-10-21
docker自动化
深入讲解Docker容器化技术的核心概念、Dockerfile编写、多阶段构建、Docker Compose使用，以及性能优化最佳实践，帮助开发者掌握现代化应用部署和环境管理技能。
2025-01-11
Mermaid 快速绘制流程图
介绍如何使用Mermaid快速绘制专业流程图，包括各种节点类型、箭头连接、子图组织和样式定制，以及在Markdown中的渲染方法，提升文档可视化效果。
2024-10-30
GitHub 协作开发指南
全面的Git和GitHub协作开发教程，涵盖开源项目贡献流程、分支管理策略、提交规范、子模块使用、PR创建与合并等，为开发者提供完整的版本控制和团队协作指南。
Comments
