Python 使用python-kafka类库开发kafka生产者客户端 (3)

consumer = KafkaConsumer('MY_TOPIC1',
                         bootstrap_servers=['127.0.0.1:9092'],
                         #auto_offset_reset='',
                         auto_offset_reset='latest',# 消费kafka中最近的数据，如果设置为earliest则消费最早的数据，不管这些数据是否消费
                         enable_auto_commit=True, # 自动提交消费者的offset
                         auto_commit_interval_ms=3000, ## 自动提交消费者offset的时间间隔
                         group_id='MY_GROUP1',
                         consumer_timeout_ms= 10000, # 如果10秒内kafka中没有可供消费的数据，自动退出
                         client_id='consumer-python3'
                         )

 

for msg in consumer:

    print (msg)

    print('topic: ', msg.topic)

    print('partition: ', msg.partition)

    print('key: ', msg.key, 'value: ', msg.value)

    print('offset:', msg.offset)

    print('headers:', msg.headers)

 

# Get consumer metrics

metrics = consumer.metrics()

print(metrics)

 

运行效果

 

 

 

通过assign、subscribe两者之一为消费者设置消费的主题

consumer = KafkaConsumer(bootstrap_servers=['127.0.0.1:9092'],

                         auto_offset_reset='latest',

                         enable_auto_commit=True, # 自动提交消费数据的offset

                         consumer_timeout_ms= 10000, # 如果1秒内kafka中没有可供消费的数据，自动退出

                         value_deserializer=lambda m: json.loads(m.decode('ascii')), #消费json 格式的消息

                         client_id='consumer-python3'

                         )

 

 

# consumer.assign([TopicPartition('MY_TOPIC1', 0)])

# msg = next(consumer)

# print(msg)

 

consumer.subscribe('MY_TOPIC1')

for msg in consumer:

    print (msg)

 

 

API及常用参数说明：

class kafka.KafkaConsumer(*topics, **configs)

*topics (str) – 可选，设置需要订阅的topic，如果未设置，需要在消费记录前调用subscribe或者assign。

 

client_id (str) – 客户端名称，默认值: ‘kafka-python-{version}’

 

group_id (str or None) – 消费组名称。如果为None，则通过group coordinator auto-partition分区分配，offset提交被禁用。默认为None

 

auto_offset_reset (str) – 重置offset策略: 'earliest'将移动到最老的可用消息, 'latest'将移动到最近消息。 设置为其它任何值将抛出异常。默认值：'latest'。

 

enable_auto_commit (bool) –  如果为True，将自动定时提交消费者offset。默认为True。

 

auto_commit_interval_ms (int) – 自动提交offset之间的间隔毫秒数。如果enable_auto_commit 为true，默认值为： 5000。

 

value_deserializer(可调用对象) - 携带原始消息value并返回反序列化后的value

 

subscribe(topics=(), pattern=None, listener=None)

订阅需要的主题

topics (list) – 需要订阅的主题列表

pattern (str) – 用于匹配可用主题的模式，即正则表达式。注意：必须提供topics、pattern两者参数之一，但不能同时提供两者。

 

metrics(raw=False)

获取消费者性能指标。

 

参考API：https://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html

 

客户端

#-*- encoding:utf-8 -*-

 

__author__ = 'shouke'

 

from kafka.client import KafkaClient

 

client = KafkaClient(bootstrap_servers=['127.0.0.1:9092'], request_timeout_ms=3000)

 

# 获取所有broker

brokers = client.cluster.brokers()

for broker in brokers:

    print('broker: ', broker)

    print('broker nodeId: ', broker.nodeId)

 

# 获取主题的所有分区

topic = 'MY_TOPIC1'

partitions = client.cluster.available_partitions_for_topic(topic)

print(partitions)

 

partition_dict = {}

partition_dict[topic] = [partition for partition in partitions]

print(partition_dict)

 

 

运行结果：

broker:  BrokerMetadata(nodeId=0, host='127.0.0.1', port=9092, rack=None)

broker nodeId:  0

{0}

{'MY_TOPIC1': [0]}

 

API及常用参数说明：

class kafka.client.KafkaClient(**configs)