这篇文章主要介绍kafka-python怎么获取topic lag值方式，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！
直接上代码吧
计算
安全
数据库
网络和加速
企业服务
from kafka import SimpleClient, KafkaConsumer
from kafka.common import OffsetRequestPayload, TopicPartition
def get_topic_offset(brokers, topic):
"""
获取一个topic的offset值的和
"""
client = SimpleClient(brokers)
partitions = client.topic_partitions[topic]
offset_requests = [OffsetRequestPayload(topic, p, -1, 1) for p in partitions.keys()]
offsets_responses = client.send_offset_request(offset_requests)
return sum([r.offsets[0] for r in offsets_responses])
def get_group_offset(brokers, group_id, topic):
"""
获取一个topic特定group已经消费的offset值的和
"""
consumer = KafkaConsumer(bootstrap_servers=brokers,
group_id=group_id,
)
pts = [TopicPartition(topic=topic, partition=i) for i in
consumer.partitions_for_topic(topic)]
result = consumer._coordinator.fetch_committed_offsets(pts)
return sum([r.offset for r in result.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怎么获取topic lag值方式”这篇文章的所有内容，感谢各位的阅读！希望分享的内容对大家有帮助，更多相关知识，欢迎关注亿速云行业资讯频道！
亿速云「云服务器」，即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘，价格低至29元/月。点击查看>>
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
