Kafka 简要使用说明
概念: Partition, consumer group, rebalance
在使用中, 为了保证消费 Kafka 的数据和生产数据的次序严格一致, 一个 Topic 只设置了一个 Partition. 我没有碰到不关心时序的场景, 所有的 Topic 都只有一个 partition.
一个 Topic 可以被多个 consumer group 重复消费, 这些 consumer group 之间是完全隔离的. 一个 consumer group 可以包含一个或多个 consumer, 但消费者个数不应该多于 partition 个数. 考虑到我们的 topic 只有一个 partition, 所以每个consumer group也只能有一个消费者.
rebalance: 如果 partition 数量(供给侧)不变, consumer group下的 consumer 数量不变, Kafaka 是不会做 rebalance 的. 但如果发生如下情形, Kafaka 就要为 consumer 重新分配 partition :
- partition 数量改变.
- consumer group 中新增了 consumer.
- Kafka broker 检测到某个 consumer 已经没有心跳, 相当于 consumer 数量减少
新建 group 如何消费完整的数据?
经常碰到一个场景是, 因为测试或其他需要, 我们想重新消费一个 topic 数据, 操作很简单, 为消费者设定一个新的 consumer group id, 同时设置 auto.offset.reset 为 earliest 即可.
为什么需要将 auto.offset.reset 特意设置为 earliest 呢? 因为, 默认的设置是 latest, 对于一个新的 group id, Kafka 在 ZooKeeper 中找不到它之前的 offset, 所有历史的数据都无法被消费, 只有新的数据才会被消费到; 设置为 earliest 就能从最早的 offset 开始消费, Kafka 默认会清空7天前的数据, 也就是说能消费最近7天的所有数据.
几个重要的属性
| 参数 | 含意 | Kafka 缺省值 | Kettle Kafka consumer推荐值 |
|---|---|---|---|
| topic | 要消费哪个topic的数据 | ||
| group.id | 消费者端自行设定一个group id | ||
| max consumption duration | 消费者持续消费Kafka时长 | 180 000 | |
| session.timeout.ms | 如果 broker 超过 session.timeout.ms 设定值还没有收到消费者的心跳汇报, 就认为消费者已经不在线, 之后 broker 安排做 consumer 之间的 rebalance | 10 000 | 30 0000 |
| heartbeat.interval.ms | 消费者主动向 broker 汇报存活的间隔时长, 取值应该是<= session.timeout.ms的1/3 | 3 000 | 3 000 |
| request.timeout.ms | 消费者消费数据的请求超时值 | 30 000 | 400 000 |
| auto.offset.reset | 消费者如何读取 offset, 有三个取值 earliest, latest, none | latest | 按需 |
| enable.auto.commit | 如果需要 Kafaka 自动更新消费的 offset(周期性更新), 设置为 true, 如果由消费者自行更新的话, 可以设置为 false. 推荐使用自动提交, 可以结合 auto.commit.interval.ms 不断调优; 如果是手动提交的话, 通常仅仅在批次消费完毕后才提交, 吞吐量是很高, 但有可能在commit 之前系统崩溃, 就会造成大量数据的重复消费. | true | |
| auto.commit.interval.ms | enable.auto.commit 开启情况下, 用来控制多久commit offset值, 间隔越大吞吐量越大, 但如果 commit 失败后重复消费的量也越大. | 5 000 |
Kafka 监控的 GUI 工具(基于 Eclipse)
网址 http://www.kafkatool.com/features.html, 个人使用免费