服务端

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

注册

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

多副本（Replica）

每个分区有多个副本。一个是leader，其他是follower，follower只能被动的向leader请求同步数据，不提供读写服务

• 虽然不支持读写分离，但kafka通过多分区做到负载均衡（每个分区leader在不同的broker上）

ISR（In-sync replicas）

只有ISR集合内的follower才有资格竞选leader，和leader保持相同的HW

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)

HW（High Watermark）

• leader的HW：决定了消费可见性，消费者只能消费HW之前的消息。leader的HW的值取所有副本通过FetchRequest上报的最小值
• 副本的HW：已同步的offset

同步机制

• 副本一直向leader发Fetch请求，如果leader有数据则立即返回，如果没有，则最多等待replica.fetch.wait.max.ms，默认500ms
• 如果leader没有数据，副本会等待fetch.max.wait.ms后再次拉取，默认500ms
• 副本在FetchRequest里上报自己的LEO和HW
  • 副本同步成功后，将在下一次FetchRequest上报自己的HW给leader，这会导致HW在一个周期内的不一致

日志

以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：缓冲区未满情况下的发送间隔

分区分配策略

• Range：平均分配
• RoundRobin：将消费组和主题分区排序，依次分配。如果消费组的信息是相同的，那么没问题
• Sticky：均衡、粘性

幂等功能（0.11.0.0）

• session级别、分区级别
• 依靠productid和seq保证，两个参数都由kafka管理
• 自动配置: acks=all,retries=Integer.MAX_VALUE,max.in.flight.requests.per.connection=1

Consumer

offset

消费的offset保存在__consumer_offsets topic下

offset的维度：每个独立的consumer或者Consumer Group，在每个topic的partition上有一个offset

• 独立consumer或者Consumer Group的group.id
• topic
• partition

反序列化

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

Consumer Group的rebalance

触发时机

• Consumer Group有组成员变更，离开或加入
• topic订阅数变化，比如使用正则订阅的
• topic partition数变化

分配策略

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

如何保证消息不丢失

producer配置

• acks=all，确保所有副本都处理成功
• retries，合理的重试次数

consumer配置

• 手动提交。处理成功后再提交offset
• 监听rebalance，rebalance时提交当前处理成功的偏移量