如何保存消费端的消费位置
什么是offset
前面在講解partition的時(shí)候，提到過(guò)offset， 每個(gè)topic可以劃分多個(gè)分區(qū)（每個(gè)Topic至少有一個(gè)分區(qū)），同一topic下的不同分區(qū)包含的消息是不同的。每個(gè)消息在被添加到分區(qū)時(shí)，都會(huì)被分配一個(gè)offset（稱(chēng)之為偏移量），它是消息在此分區(qū)中的唯一編號(hào)，kafka通過(guò)offset保證消息在分區(qū)內(nèi)的順序，offset的順序不跨分區(qū)，即kafka只保證在同一個(gè)分區(qū)內(nèi)的消息是有序的； 對(duì)于應(yīng)用層的消費(fèi)來(lái)說(shuō)，每次消費(fèi)一個(gè)消息并且提交以后，會(huì)保存當(dāng)前消費(fèi)到的最近的一個(gè)offset。那么offset保存在哪里？
offset在哪里維護(hù)？
在kafka中，提供了一個(gè)consumer_offsets_* 的一個(gè)topic，把offset信息寫(xiě)入到這個(gè)topic中。consumer_offsets——按保存了每個(gè)consumer group某一時(shí)刻提交的offset信息。__consumer_offsets 默認(rèn)有50個(gè)分區(qū)。
根據(jù)前面我們演示的案例，我們?cè)O(shè)置了一個(gè)KafkaConsumerDemo的groupid。首先我們需要找到這個(gè)consumer_group保存在哪個(gè)分區(qū)中
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"KafkaConsumerDemo");
計(jì)算公式
Math.abs(“groupid”.hashCode())%groupMetadataTopicPartitionCount ; 由于默認(rèn)情況下groupMetadataTopicPartitionCount有50個(gè)分區(qū)，計(jì)算得到的結(jié)果為:35, 意味著當(dāng)前的consumer_group的位移信息保存在__consumer_offsets的第35個(gè)分區(qū)
執(zhí)行如下命令，可以查看當(dāng)前consumer_goup中的offset位移提交的信息
kafka-console-consumer.sh --topic __consumer_offsets --partition 15 -- bootstrap-server 192.168.13.102:9092,192.168.13.103:9092,192.168.13.104:9092 --formatter 'kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter'從輸出結(jié)果中，我們就可以看到test這個(gè)topic的offset的位移日志
?
?
?
?
?
?
?
?
?
?
?
總結(jié)
以上是生活随笔為你收集整理的如何保存消费端的消费位置的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。
