python kafka client--confluent-kafka-python

谁践踏了优雅 2022-05-23 13:41 215阅读 0赞

项目中需要使用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][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][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。

[https_github.com_confluentinc_confluent-kafka-python]: https://github.com/confluentinc/confluent-kafka-python
[http_docs.confluent.io_current_installation.html_rpm-packages-via-yum]: http://docs.confluent.io/current/installation.html#rpm-packages-via-yum