python 操作Kafka kafka-python :消费者学习

1.1 安装

> pip install kafka-python

1.2 消费者示例

# coding:utf8
from kafka import KafkaConsumer
#创建一个消费者,指定了topic,group_id,bootstrap_servers
#group_id:多个拥有相同group_id的消费者被判定为一组,一条数据记录只会被同一个组中的一个消费者消费
#bootstrap_servers:kafka的节点,多个节点使用逗号分隔
#这种方式只会获取新产生的数据

consumer = KafkaConsumer(
    bootstrap_servers = "192.168.70.221:19092,192.168.70.222:19092,192.168.70.223:19092", # kafka集群地址
    group_id = "my.group", # 消费组id
    enable_auto_commit = True, # 每过一段时间自动提交所有已消费的消息(在迭代时提交)
    auto_commit_interval_ms = 5000, # 自动提交的周期(毫秒)
)

consumer.subscribe(["my.topic"]) # 消息的主题,可以指定多个

for msg in consumer: # 迭代器,等待下一条消息
    print msg # 打印消息

1.3 KafkaConsumer的构造参数:

  • *topics ,要订阅的主题
  • bootstrap_servers :kafka节点或节点的列表,不一定需要罗列所有的kafka节点。格式为: ‘host[:port]’ 。默认值是:localhost:9092
  • client_id (str) : 客户端id,默认值: ‘kafka-python-{version}’
  • group_id (str or None):分组id
  • key_deserializer (callable) :key反序列化函数
  • value_deserializer (callable):value反序列化函数
  • fetch_min_bytes:服务器应每次返回的最小数据量
  • fetch_max_wait_ms (int): 服务器应每次返回的最大等待时间
  • fetch_max_bytes (int) :服务器应每次返回的最大数据量
  • max_partition_fetch_bytes (int) :
  • request_timeout_ms (int) retry_backoff_ms (int)
  • reconnect_backoff_ms (int)
  • reconnect_backoff_max_ms (int)
  • max_in_flight_requests_per_connection (int)
  • auto_offset_reset (str) enable_auto_commit (bool)
  • auto_commit_interval_ms (int)
  • default_offset_commit_callback (callable)
  • check_crcs (bool)
  • metadata_max_age_ms (int)
  • partition_assignment_strategy (list)
  • max_poll_records (int)
  • max_poll_interval_ms (int)
  • session_timeout_ms (int)
  • heartbeat_interval_ms (int)
  • receive_buffer_bytes (int)
  • send_buffer_bytes (int)
  • socket_options (list)
  • consumer_timeout_ms (int)
  • skip_double_compressed_messages (bool)
  • security_protocol (str)
  • ssl_context (ssl.SSLContext)
  • ssl_check_hostname (bool)
  • ssl_cafile (str) –
  • ssl_certfile (str)
  • ssl_keyfile (str)
  • ssl_password (str)
  • ssl_crlfile (str)
  • api_version (tuple)

1.4 KafkaConsumer的函数

  • assign(partitions):手动为该消费者分配一个topic分区列表。
  • assignment():获取当前分配给该消费者的topic分区。
  • beginning_offsets(partitions):获取给定分区的第一个偏移量。
  • close(autocommit=True):关闭消费者
  • commit(offsets=None):提交偏移量,直到成功或错误为止。
  • commit_async(offsets=None, callback=None):异步提交偏移量。
  • committed(partition):获取给定分区的最后一个提交的偏移量。
  • end_offsets(partitions):获取分区的最大偏移量
  • highwater(partition):分区最大的偏移量
  • metrics(raw=False):返回消费者性能指标
  • next():返回下一条数据
  • offsets_for_times(timestamps):根据时间戳获取分区偏移量
  • partitions_for_topic(topic):返回topic的partition列表,返回一个set集合
  • pause(*partitions):停止获取数据paused():返回停止获取的分区poll(timeout_ms=0, max_records=None):获取数据
  • position(partition):获取分区的偏移量
  • resume(*partitions):恢复抓取指定的分区
  • seek(partition, offset):seek偏移量
  • seek_to_beginning(*partitions):搜索最旧的偏移量
  • seek_to_end(*partitions):搜索最近可用的偏移量
  • subscribe(topics=(), pattern=None, listener=None):订阅topics
  • subscription():返回当前消费者消费的所有topic
  • topics():返回当前消费者消费的所有topic,返回的是unicode
  • unsubscribe():取消订阅所有的topic

1.5 高级用法(消费者)

从指定offset开始读取消息,被消费过的消息也可以被此方法读取

  1. 创建消费者
    2.使用 assign 方法重置指定分区(partition)的读取偏移(fetch offset)的值
  2. 使用 seek 方法从指定的partition和offset开始读取数据
#encoding:utf8
from kafka import KafkaConsumer, TopicPartition

my_topic = "my.topic" # 指定需要消费的主题

consumer = KafkaConsumer(
    bootstrap_servers = "192.168.70.221:19092,192.168.70.222:19092,192.168.70.223:19092", # kafka集群地址
    group_id = "my.group", # 消费组id
    enable_auto_commit = True, # 每过一段时间自动提交所有已消费的消息(在迭代时提交)
    auto_commit_interval_ms = 5000, # 自动提交的周期(毫秒)
)

consumer.assign([
    TopicPartition(topic=my_topic, partition=0),
    TopicPartition(topic=my_topic, partition=1),
    TopicPartition(topic=my_topic, partition=2)
])

consumer.seek(TopicPartition(topic=my_topic, partition=0), 12) # 指定起始offset为12
consumer.seek(TopicPartition(topic=my_topic, partition=1), 0) # 可以注册多个分区,此分区从第一条消息开始接收
# consumer.seek(TopicPartition(topic=my_topic, partition=2), 32) # 没有注册的分区上的消息不会被消费

for msg in consumer: # 迭代器,等待下一条消息
    print msg # 打印消息

## 其他用法

#立刻发送所有数据并等待发送完毕
producer.flush()

#读取下一条消息
next(consumer)

#手动提交所有已消费的消息
consumer.commit()

#手动提交指定的消息
consumer.commit([
    TopicPartition(my_topic, msg.offset)
])

1.6 KafkaProductor的工具类

# -*- coding: utf-8 -*-

"""
Create by Mr.Hao on 2019/7/12.

"""

import sys
import json
import traceback
from utils import get_logger
from kafka.errors import KafkaError
from kafka import KafkaConsumer,TopicPartition
logger = get_logger("ConsumerForKFK")


class ConsumerForKFK(object):


    _MESSAGE_NAME = ''

    def __init__(self, kafkahost, client_id):
        self.kafkaHost = kafkahost
        self.client_id = client_id

    @property
    def consumer_client(self, group_id=None):
        return KafkaConsumer(self._MESSAGE_NAME,
                      bootstrap_servers=self.kafkaHost,
                      client_id = self.client_id,
                      """
                      # earliest
                      当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
                      # latest
                       当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
                      # none
                      topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常
                      """
                      auto_offset_reset='earliest',
                      group_id=group_id,
                      # 若不指定 consumer_timeout_ms,默认一直循环等待接收,若指定,则超时返回,不再等待
                      # consumer_timeout_ms : 毫秒数
                      consumer_timeout_ms=5000)
    
    @property
    def close(self):
        return self.consumer_client.close()

    def consumer(self):
        """
        :return: 返回正常的kfk 消息
        """
        try:
            consumer = self.consumer_client
            # 迭代器,等待下一条消息
            for msg in consumer:
                # 打印消息
                print msg
        except KafkaError as e:
            t, v, tb = sys.exc_info()
            logger.error("send msg ext has error, please check: %s, %s, %s", t, v, traceback.format_tb(tb))
        finally:
            self.close()

    def consumer_seek(self, partition=1, offset=0):
        """
        :param partition: 
        :param offset: 
        :return: 
        """
        try:
            consumer = self.consumer_client

            consumer.seek(TopicPartition(topic=self._MESSAGE_NAME, partition=partition), offset)
            # 发送到指定的消息主题(异步,不阻塞)
            for msg in consumer:  # 迭代器,等待下一条消息
                print msg  # 打印消息

        except KafkaError as e:
            t, v, tb = sys.exc_info()
            logger.error("send msg ext has error, please check: %s, %s, %s", t, v, traceback.format_tb(tb))
        finally:
            self.close()

    def consumer_assign(self, partition=1):
        """
        :param partition: 
        :return: 
        """
        try:
            consumer = self.consumer_client
            consumer.assign([TopicPartition(topic=self._MESSAGE_NAME, partition=partition)])
            # 发送到指定的消息主题(异步,不阻塞)
            for msg in consumer:  # 迭代器,等待下一条消息
                print msg  # 打印消息
        except KafkaError as e:
            t, v, tb = sys.exc_info()
            logger.error("send msg ext has error, please check: %s, %s, %s", t, v, traceback.format_tb(tb))
        finally:
            self.close()
文档

kafka 文档
kafka-python
kafka-python 文档

文章:用管道解释kafka
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 204,189评论 6 478
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,577评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,857评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,703评论 1 276
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,705评论 5 366
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,620评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,995评论 3 396
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,656评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,898评论 1 298
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,639评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,720评论 1 330
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,395评论 4 319
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,982评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,953评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,195评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 44,907评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,472评论 2 342