python?實時獲取kafka消費隊列信息示例詳解

更新時間：2023年07月24日 08:40:03 作者：xiaoming0018

這篇文章主要介紹了python實時獲取kafka消費隊列信息,本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下

一、消費kafka消息

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from pykafka import KafkaClient
from pykafka.common import OffsetType
from vpn_data_handler import handler_data
bootstrap_servers = '10.*.**.**:9092'
group_id = 'test1'
class KConsumer(object):
    """kafka 消費者; 動態(tài)傳參,非配置文件傳入;
      kafka 的消費者應該盡量和生產者保持在不同的節(jié)點上;否則容易將程序陷入死循環(huán)中;
     """
    _encode = "UTF-8"
    def __init__(self, topics, bootstrap_server=None, group_id=group_id, partitions=None):
        """ 初始化kafka的消費者;
           1. 設置默認 kafka 的主題, 節(jié)點地址, 消費者組 id(不傳入的時候使用默認的值)
           2. 當需要設置特定參數的時候可以直接在 kwargs 直接傳入,進行解包傳入原始函數;
         Args:
           topics: str; kafka 的消費主題;
           bootstrap_server: list; kafka 的消費者地址;
           group_id: str; kafka 的消費者分組 id,默認是 start_task 主要是接收并啟動任務的消費者,僅此一個消費者組id;
         """
        if bootstrap_server is None:
            bootstrap_server = bootstrap_servers
        self.client = KafkaClient(hosts=bootstrap_server)
        # 選擇要消費的topic
        vpn_topic = self.client.topics[topics]
        self.consumer = vpn_topic.get_simple_consumer(consumer_group=group_id,
                                                      consumer_timeout_ms=200,
                                                      auto_commit_enable=True,# 自動提交偏移量
                                                      auto_offset_reset=OffsetType.LATEST)  #LATEST 獲取當前偏移量最新消息  EARLIEST從頭開始獲取信息
    def recv(self):
        """
         接收消費中的數據
         Returns:
         """
        return self.consumer
def main():
    """
    kafka消費隊列入口
    :param topic:
    :return:
    """
    obj = KConsumer(topics="topics_name")
    while True:
        for message in obj.recv():
            data = eval(message.value.decode('utf-8'))
            handler_data(data)
if __name__ == '__main__':
    main()

二、生產者推送消息

#!/usr/bin/python
# -*- coding:utf-8 -*-
from pykafka import KafkaClient
client = KafkaClient(hosts="10.XX0.XX0.XX4:9092")  # 可接受多個client
# 查看所有的topic
# print(client.topics)
topic = client.topics['test_78'] # 選擇一個topic
message = "test message2 test message2"
with topic.get_sync_producer() as producer:
    producer.produce(bytes(message, encoding='utf8')) #python3需要編碼
    print(message)

到此這篇關于python 實時獲取kafka消費隊列信息的文章就介紹到這了,更多相關python kafka消費隊列信息內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: