浅入浅出kafka

弗兰兹·卡夫卡，生活于奥匈帝国统治下的捷克德语小说家，本职为保险业职员。主要作品有小说《审判》、《城堡》、《变形记》等。

another

Kafka是由Apache软件基金会开发的一个开源平台，由Scala和Java编写。Kafka是一种的，它可以处理消费者规模的网站中的所有动作流数据。

两个kafka有一个共同特别: 很会写

1. 消息系统

低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能，成为异步RPC的主要手段之一。

1.1. 什么是消息系统

producer发送消息给broker，broker持有数据，在合适的时机发送给consumer，consumer确认后，broker删除消息数据。

优化点无外乎吞吐量、性能、可靠性、事务。

1.1.1 扩展概念

消息队列模型 sender -> queue -> receiver p2p

发布/订阅模型 publish -> topic -> subscribe

push

pull

1.2. 什么时候使用消息系统

1.2.1. 适合场景

业务解耦，领域更清晰。区分业务核心系统

最终一致性(反之，强一致性，需要接收方回调确认，同步RPC更合适)

广播，1 VS N，稳定上游服务

错峰流控，拉平峰值，避免木桶

日志同步

1.2.2. 不适合场景

强事务保证

延迟敏感，实时响应

2. kafka好在哪

2.1. 吞吐量/延时

吞吐量: 每秒能够处理的消息or字节数。

延时: 客户端发送请求、服务端处理请求并发送相应给客户端。

延时越低，吞吐越高？

通常情况下，我们认为延时越低，单位时间可以处理的请求变多，所以吞吐量增加。但是两者并不是正相关关系。

e.g. kafka处理一条消息需要花费2ms，吞吐量为1000/2=500。如果通过batch，批量发送，每8s发送一次600条，延时=2ms+8ms=10ms，600*(1000/10)=60000。

2.2. 消息持久化

保存在硬盘，不会丢失，可以重放。and性能很高!!!后面聊原因。

2.3. 负载均衡和故障转移

多副本、多分区，保障高可用。

2.4. 伸缩性

自身无状态，方便扩展。

3. 名词解析

message -> 消息

broker -> kafka服务器

topic -> 主题，逻辑概念，一类消息，一个消息内容体

partition -> 分区，消息实际存储的物理位置。有序队列，维护offset。

同一个topic可以在不同broker上维护不同的分区(负载均衡)

同一个topic可以在不同broker上维护同一个分区(冗余机制,故障转移)

replica -> 副本(partition)。分为leader replica 和 follower replica。和Master-Slave不同，follower只从leader同步数据，不提供读写。只有在leader挂了之后，才会选举follower作为leader提供服务。kafka保障同一个partition的replica在不同的broker，否则无法提供故障转移。同一个topic可以有不同的leader，同一个topic+partition只有一个leader。

ISR(is-sync replica) -> 同步副本集合。如果follower延迟过大，会被踢出集合，追赶上数据之后，重新向leader申请，加入ISR集合。并不是所有的follower都可以成为leader，ISR集合中的follower可以竞选leader。通过replica.lag.time.max.ms(默认10s)设置follower同步时间，通过RetchRequest(offset)同步leader信息。

offset -> 位移、偏移量

上次提交的位移:group确认的offset

当前位置:读取后，未提交

HW:ISR确认已同步后，leader增加HW。

LEO: leader接收到的最新一条producer发送的数据

consumer只能消费到HW，未同步给所有ISR成员的消息无法消费

leader保存 LEO、HW和remote LEO, min(LEO, remote LEO) 更新HW

follower轮询leader，purgatory暂存请求，500ms

新版本epoch保存leader变更版本，维护kv (epoch, offset)

producer -> 生产者

consumer -> 消费者

group -> 组。通过维护各group的offset，每条消息只会被发送到同一个group下的一个consumer，实现不同模型。

一个group有一或多个consumer

一个消息可以发送给多个group

controller -> 控制器。选举broker作为controller，管理和协调kafka集群，和zookeeper通信。

coordinator -> 协调者。用于实现成员管理、消费分配方案制定(rebalance)以及提交位移等，每个group选举consumer作为协调者。

4. kafka高性能的秘密

4.1. 顺序写?

网上的教程经常看到介绍，写入耗时主要集中在磁头寻道和盘片旋转，而数据传输速度很快。kafka采用了顺序写，所以效率高。不免有些疑问:

顺序写性能高，为什么还有随机写？

磁盘不会被占用，每次写入都需要寻道、旋转，那么顺序写的优势在哪？

原因

因为写入的是不同的文件，占用连续的page。顺序写，不能修改。

增加前提:一次写入一个文件且文件足够大。

所以本质原因在于追加写，"每个partition是一个文件"。

读取时，识别顺序写，会进行预读。

4.2. PageCache

Kafka不会每次都写磁盘，而是写入分页存储PageCache就认为producer成功。

操作系统决定什么时候将PageCache写入磁盘(flush)。增加flush时间间隔，可以提升吞吐

flush时为顺序写入，不会有额外的性能损耗。

读取时，优先读取PageCache。

4.3. PageCache为缓存，数据会不会丢失？

因为是操作系统管理，所以kafka进程挂了，数据不会丢失。如果操作系统掉电。。。依靠副本

4.4. Zero Copy

java FileChannel.transferTO

linux sendfile

5. partition

leader针对partition而不是broker

partition不是一个文件而是一个文件夹

partition是我们能操作的最小概念

如果一直追加会导致文件过大，不便于使用(读写)和维护(删除旧数据)，kafka为此采用了几种措施

区分segment

增加索引,包括index和timeindex

5.1. segment

5.2. index和timeindex

索引文件预分配空间，切分时裁剪。

p.s. producer发送消息时，可以指定时间戳。如果机器时区不同，或者retry、网络延时等导致时间混乱，按照时间索引进行查询时，导致查询不到消息。?? 时间会在发送时获取本机时间

6. producer

6.1. producer配置

通过配置文件了解细节

bootstrap.servers 指定其中一个，会自动找到leader，但是如果指定的机器挂了，无法切换

acks 0, 1, all|-1。 0表示无需确认，1表示leader确认，-1表示所有ISR确认。

buffer.memory 缓存消息的缓冲区大小32MB，过小会影响吞吐。写入速度超过发送速度，停止&等待IO发送，still追不上会报错。

compression.type 开启压缩，提升IO吞吐，增加CPU压力。需要看服务器是IO密集型or计算密集型。 属性 0: 无压缩，1: GZIP，2: Snappy，3: LZ4

retries 重试，屏蔽网络抖动or leader选举 or NotController，导致消息重复发送。详细参见

retry.backoff.ms 重试间隔

batch.size 批量发送大小，默认16KB，增加可提升吞吐

linger.ms 发送时间，默认为0，立即发送，不判断batch.size大小。

max.request.size 消息大小，因为存在header等，实际大小大于消息本身

request.timeout.ms 超时时间，默认30s。broker给producer反馈

partitioner.class

key.serializer & value.serializer

interceptor.classes 自定义拦截器

6.2. 自定义serializer

6.3. 自定义partitioner

自定义分区策略 机器人发送到同一个partition，为了快速响应真实用户。如果只是为了均匀分布，不需要指定key(和旧版本不同)。

如果未指定key，会通过轮询，避免skewed

6.4. producer拦截器

6.5. 100% 送达配置 | 无消息丢失配置

通过配置替换

6.6. 消息内容7. Consumer

为什么不用zookeeper保存？

zookeeper不擅长频繁写(强一致性)

为什么不用broker保存？

增加应答机制，确认消费成功，影响吞吐

保存多个consumer的offset，数据结构复杂

7.1. Consumer配置7.2. partition分配策略

每个partition分配给一个consumer。

e.g. 如果一个group订阅一个topic，一个topic有100个partition，一个group有5个consumer。则每个consumer消费20个partition

partition分配策略，继承AbstractPartitionAssignor自定义策略规则，加权重等。自带分配规则:

range 分区顺序排列、分组、分配给consumer

round-robin 分区顺序排列， 轮询consumer，读取分区

sticky 基于历史分配方案，避免数据倾斜

7.3. 误解

使用过程中对kafka consumer的一些误解

7.3.1. 误解1

和按照时间或者消息记录数，控制每次获取消息。

poll表示轮询，使用poll而不是pull，并不需要wakeup。所以可以使用，每次数据流准备好后，会返回并进行业务处理。

7.3.2. 误解2

"consumer只能订阅一个topic。"

7.3.3. 误解3

"commitSync同步提交，阻塞消费。commitAsync异步提交，不阻塞消费。"

commitSync和commitAsync都会阻塞poll，因为在poll执行时轮询时会判断commit状态。commitAsync不阻塞业务处理后续方法执行。

7.4. EOS(Exactly-once Semantics)

at most once 最多一次，消息可能丢失，但不会被重复处理。获取消息后，先commit，然后业务处理。

at least once 最少一次 消息不会丢失，但可能被处理多次。获取消息后，先业务处理，然后commit。

exactly once 会被处理且只会被处理一次

7.5. 消费指定partition

自定义分配策略？不需要，可以通过assign指定topic partition

assign + subscribe 冲突错误

assign + assign 后一个生效

2个consumer assign同一个partition 消费两次

一个consumer assign 一个consumer subscribe, rebalance 踢出assign

7.6. 控制提交分区offset

Mapoffsets 控制提交分区offset，细粒度

7.7. rebalance

状态机

触发条件

consumer加入、退出、崩溃

topic发生变更，如正则匹配，增加topic

分区发生变动

消费处理超时

rebalance generation 标识 rebalance，每次+1， 延迟提交offset会被group拒绝 ILLEGAL_GENERATION

协议

joinCroup 请求

SyncGroup 请求，group leader 同步分配方案

Heartbeat 请求 向coordinator汇报心跳

LeaveGroup 请求

DescribeGroup 查看组信息

收集join consumer，选取leader，同步给coordinator。 leader 负责分配

同步更新分配方案，发送SyncGroup请求给coordinator，每个consumer都发送，coordinator接受leader的方案，分配，返回response

7.8. 多线程消费

自己实现缓存区，批量执行及确认consumer.commitSync

多consumer thread 效率高，单独offset。 缺点：受限于topic分区数，broker压力大，rebalance可能性大

单consumer 多handler thread ，获取和处理节藕，伸缩性好。难于管理分区内消息顺序，位移提交困难，处理不当导致数据丢失。

8. 其他

8.1. 日志留存8.2. 暂停consumer消费

e.g. 消费逻辑为调用三方接口，如果三方接口不稳定，需要关闭一段时间。

暂停

启动

8.3. compaction

订阅binlog seecanal

高可用日志化

8.4. manager指定lisnterners

因为kafka内部使用全称域名，不统一，导致无法获取元数据

8.5. 生产环境

优雅的启动和关闭(Spring生命周期)

offset跳过与重放

附录

Apache Kafka实战

Page Cache的落地问题

Kafka文件存储机制那些事

Netty 之 Zero-copy 的实现（下）

磁盘I/O那些事

Kafka消费组(consumer group)