python获取kafka队列长度

# pip install kafka-python

from kafka import KafkaConsumer
from kafka.structs import TopicPartition

def get_queue_length(topic_list, kafka_host, kafka_group):

    partition_list = [TopicPartition(it[1], 0) for it in topic_list]
    consumer = KafkaConsumer(bootstrap_servers=[kafka_host], group_id=kafka_group)
    consumer.assign(partition_list)
    off_set_dict = consumer.end_offsets(partition_list)
    latest_offset = list(off_set_dict.values())
    cursor_li = list(map(lambda x: consumer.position(x), partition_list))
    queue_len = sum(latest_offset) - sum(cursor_li)
    return queue_len


if __name__ == "__main__":
    # 订阅主题列表
    topic_list = ['test', ]
    # 消费组
    kafka_group = 'test_group'
    # 连接地址
    kafka_host = '127.0.0.1:9092'
    q_length = get_queue_length(topic_list, kafka_host, kafka_group)
    print(q_length)