kafka如何保存消费端的消费位置

时间 2019-12-21

标签 kafka 如何保存消费端的位置栏目 Kafka 繁體版

原文原文链接

在kafka 中，提供了一个 ___consumer_ offsets_* 的一个内置topicbootstrap

__consumer_offsets 默认分区50
建立consumer_offsets_group 消费组ID 稍后会用到
- 经过以下公式便可获取：
  - Math.abs("consumer_offsets_group".hashCode()) % 50
  - 结果放在了11分区上
查看该消费群offsets：
- bin/kafka-console-consumer.sh --topic __consumer_offsets --partition 11 --bootstrap-server localhost:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --consumer.config config/consumer.properties --from-beginning
结果：
- [consumer_offsets_group/*消费组ID*/,consumer_offsets_test/*topic*/,1/*partition*/]::OffsetAndMetadata(offset=4/*提交的消费位移信息*/, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1543567461031, expireTimestamp=None) [consumer_offsets_group,consumer_offsets_test,0]::OffsetAndMetadata(offset=3, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1543567461031, expireTimestamp=None) [consumer_offsets_group,consumer_offsets_test,2]::OffsetAndMetadata(offset=4, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1543567461031, expireTimestamp=None) [consumer_offsets_group,consumer_offsets_test,1]::OffsetAndMetadata(offset=4, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1543567466027, expireTimestamp=None) [consumer_offsets_group,consumer_offsets_test,0]::OffsetAndMetadata(offset=3, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1543567466027, expireTimestamp=None) [consumer_offsets_group,consumer_offsets_test,2]::OffsetAndMetadata(offset=4, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1543567466027, expireTimestamp=None) [consumer_offsets_group,consumer_offsets_test,1]::OffsetAndMetadata(offset=4, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1543567468638, expireTimestamp=None) [consumer_offsets_group,consumer_offsets_test,0]::OffsetAndMetadata(offset=3, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1543567468638, expireTimestamp=None) [consumer_offsets_group,consumer_offsets_test,2]::OffsetAndMetadata(offset=4, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1543567468638, expireTimestamp=None)