kafka-python 获取topic lag值方式
说真，这个问题看上去很简单，但“得益”与kafka-python神奇的文档，真的不算简单，反正我是搜了半天还看了半天源码。
直接上代码吧
fromkafkaimportSimpleClient,KafkaConsumer
fromkafka.commonimportOffsetRequestPayload,TopicPartition
defget_topic_offset(brokers,topic):
"""
获取一个topic的offset值的和
"""
client=SimpleClient(brokers)
partitions=client.topic_partitions[topic]
offset_requests=[OffsetRequestPayload(topic,p,-1,1)forpinpartitions.keys()]
offsets_responses=client.send_offset_request(offset_requests)
returnsum([r.offsets[0]forrinoffsets_responses])
defget_group_offset(brokers,group_id,topic):
"""
获取一个topic特定group已经消费的offset值的和
"""
consumer=KafkaConsumer(bootstrap_servers=brokers,
group_id=group_id,
)
pts=[TopicPartition(topic=topic,partition=i)foriin
consumer.partitions_for_topic(topic)]
result=consumer._coordinator.fetch_committed_offsets(pts)
returnsum([r.offsetforrinresult.values()])
if__name__=='__main__':
topic_offset=get_topic_offset("brokers","topic")
group_offset=get_group_offset("brokers","group_id","topic")
lag=topic_offset-group_offset
以上这篇kafka-python获取topiclag值方式就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持毛票票。
声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。