python 实时获取kafka消费队列信息示例详解

 更新时间:2023年07月24日 08:40:03   作者:xiaoming0018  
这篇文章主要介绍了python实时获取kafka消费队列信息,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下

安装 pykafka

pip install pykafka

一、消费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 消费者; 动态传参,非配置文件传入;
      kafka 的消费者应该尽量和生产者保持在不同的节点上;否则容易将程序陷入死循环中;
     """
    _encode = "UTF-8"
    def __init__(self, topics, bootstrap_server=None, group_id=group_id, partitions=None):
        """ 初始化kafka的消费者;
           1. 设置默认 kafka 的主题, 节点地址, 消费者组 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消费队列信息内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • 基于Python实现贪吃蛇小游戏(附源码)

    基于Python实现贪吃蛇小游戏(附源码)

    本次我们将编写一个贪吃蛇的游戏。通过键盘上、下、左、右控制小蛇上、下、左、右移动,吃到食物后长度加1;蛇头碰到自身或窗口边缘,游戏失败,需要的可以参考一下
    2022-11-11
  • 使用Python+Splinter自动刷新12306火车票

    使用Python+Splinter自动刷新12306火车票

    一年一度的春运又来了,今年我自己写了个抢票脚本,下面小编给大家分享使用Python+Splinter自动刷新12306火车票,需要的朋友参考下吧
    2018-01-01
  • 对Python之gzip文件读写的方法详解

    对Python之gzip文件读写的方法详解

    今天小编就为大家分享一篇对Python之gzip文件读写的方法详解,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2019-02-02
  • Python tkinter库图形绘制例子分享

    Python tkinter库图形绘制例子分享

    这篇文章主要介绍了 Python tkinter库图形绘制例子,文章基于Python的相关资料展开主题,分享绘图小例子,需要的小伙伴可以参考一下
    2022-04-04
  • Pandas之缺失数据的实现

    Pandas之缺失数据的实现

    这篇文章主要介绍了Pandas之缺失数据的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2021-01-01
  • python名片管理系统开发

    python名片管理系统开发

    这篇文章主要为大家详细介绍了python名片管理系统开发,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2020-06-06
  • Python数学建模学习模拟退火算法整数规划问题示例解析

    Python数学建模学习模拟退火算法整数规划问题示例解析

    整数规划问题在工业、经济、国防、医疗等各行各业应用十分广泛,是指规划中的变量(全部或部分)限制为整数,属于离散优化问题Discrete Optimization
    2021-10-10
  • Python读取图片的方法详解

    Python读取图片的方法详解

    这篇文章主要为大家详细介绍了Python中读取图片的实现方法,文中的示例代码简洁易懂,具有一定的参考价值,需要的小伙伴可以跟随小编一起学习一下
    2023-08-08
  • 用Python读取几十万行文本数据

    用Python读取几十万行文本数据

    今天小编就为大家分享一篇关于用Python读取几十万行文本数据,小编觉得内容挺不错的,现在分享给大家,具有很好的参考价值,需要的朋友一起跟随小编来看看吧
    2018-12-12
  • python实现简单颜色识别程序

    python实现简单颜色识别程序

    这篇文章主要为大家详细介绍了python实现简单颜色识别程序,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2020-02-02

最新评论