欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

kafka-python 獲取topic lag值方式

 更新時(shí)間：2019年12月23日 09:26:33   作者：conghui_  
今天小編就為大家分享一篇kafka-python 獲取topic lag值方式，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧

說(shuō)真，這個(gè)問(wèn)題看上去很簡(jiǎn)單，但“得益”與kafka-python神奇的文檔，真的不算簡(jiǎn)單，反正我是搜了半天還看了半天源碼。

直接上代碼吧

from kafka import SimpleClient, KafkaConsumer
from kafka.common import OffsetRequestPayload, TopicPartition

def get_topic_offset(brokers, topic):
  """
  獲取一個(gè)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):
  """
  獲取一個(gè)topic特定group已經(jīng)消費(fèi)的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值方式就是小編分享給大家的全部?jī)?nèi)容了，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

相關(guān)文章

最新評(píng)論