python kafka client--confluent-kafka-python

项目中需要使用python 向Kafka生产和消费数据，最初使用pykafka 。后来发现pykafka不支持client.id.

最后，终于找到confluent-kafka。

python kafka推荐使用confluent-kafka，官方推荐的。

Confluent's Apache Kafka Python client

confluent-kafka-python is Confluent's Python client for Apache Kafka and the Confluent Platform.

安装和使用可参考github

https://github.com/confluentinc/confluent-kafka-python

本文只针对笔者使用python2.7 + Linux环境介绍kafka包安装和使用。

安装python

可网上搜索教程，此处略过。

安装librdkafka

For RedHat and RPM-based distros, add this YUM repo and then do sudo yum install librdkafka-devel python-devel: http://docs.confluent.io/current/installation.html#rpm-packages-via-yum

其他系统，可以参考github上说明。

安装 python kafka pacakge

$ pip install confluent-kafka

Successfully installed confluent-kafka-0.11.4

consumer 举例

\#!/usr/bin/python

from confluent_kafka import Consumer, KafkaError

mybroker = "127.0.0.1:9092"

c = Consumer({

    'bootstrap.servers': mybroker,

    'group.id': 'mygroup',

    'client.id': 'lanyang',

    'default.topic.config': {

    'auto.offset.reset': 'smallest'

    }

})

c.subscribe(['test'])

while True:

    msg = c.poll(1.0)

    if msg is None:

        continue

    if msg.error():

        if msg.error().code() == KafkaError._PARTITION_EOF:

            continue

        else:

            print(msg.error())

            break

    print('Received message: {}'.format(msg.value().decode('utf-8')))

c.close()

输出结果

$ python kafa_consumer.py

Received message: Hello,python

Received message: Hello,kafka

更多demo请参考github。