服务端

下载地址：https://kafka.apache.org/downloads

注册

broker向ZK注册：$chroot/brokers/ids/{broker.id}

多副本（Replica）

一个是leader，其他是follower，follower只能被动的向leader请求同步数据，不能提供服务

ISR（In-sync replicas）：只有ISR集合内的follower才有资格竞选leader

1
2
3

0  1  2  3  4  5  6  7  8  9  10  11  12  13
|                 |                               |
base offset       high watermark(HW)       log end offset(LEO)

日志

以partition为单位，每个partition有自己的日志（partition log），如果topic名为test，则其第0个分区的文件目录类似如下：

1
2
3
4
5
6

> ls xxx/kafka_1/test-0
00000.index
00000.log // 真实kafka记录
00000.timeindex
01000.log  // 文件命名为第一条记录的offset，也就是说00000.log文件里包含1000个记录（0-999）
01000.timeindex

• .log文件命名为第一条记录的offset，也就是说00000.log文件里包含1000个记录（0-999）
• .index是位移索引文件，.timeindex是时间戳索引文件，二者都是稀疏索引

日志清除

• 基于时间
• 基于大小

日志Compact

对于一个partition内的相同key的记录，仅保存最新的值，没有key的无法compact

• compact不会变更offset

Controller

某个broker会被选举成为controller，用于管理每个broker上的分区副本和每个分区的leader副本信息

Gradle配置：org.apache.kafka:kafka-clients:3.4.0

Producer

基本功能

1. 序列化
2. 分区
   • 如果没有设置partition，则轮询（peekCurrentPartitionInfo）
   • 如果有Key，则使用murmur2算法计算hash值
   • 自定义
3. 拦截器
   • Properties配置项：interceptor.classes
   • 接口：ProducerInterceptor
   • onSend：发送前调用
   • onAcknowledgement：broker返回后调用
   • close：执行一些清理动作
4. 压缩
   • GZIP
   • Snappy
   • LZ4
   • Zstd
5. Sender发送
   • 将消息放入缓冲区，由IO线程发送
   • 收到response后，依次调用回调方法

重要参数

• acks：多少副本写入成功才返回，取值-1/all，0，1
• retires：重试次数
• batch.size：同一分区的缓冲区大小
• linger.ms：缓冲区未满情况下的发送间隔

Consumer

消费的offset保存在__consumer_offsets topic下

反序列化

org.apache.kafka.clients.consumer.internals.Fetcher#parseRecord

rebalance

触发时机

• 有组成员变更，离开或加入
• topic订阅数变化
• topic分区数变化

分配策略

• range：固定范围分配
• round-robin：轮询分配
• sticky：有粘性的分配，最大限度地维持之前的分配方案