您现在的位置是：首页 > Python

当前栏目

python kafka client--confluent-kafka-python

Python GitHub

2023-02-26 12:32:23 时间

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

猜你喜欢

最长无重复子串
写技术博客的一些心得分享
Java 多线程（七）：线程池
Java 多线程（五）：锁（三）
Java 多线程（四）：锁（二）
Java 多线程（三）：锁（一）
Java 多线程（二）：并发编程的三大特性
线性时间非比较类排序
Java 多线程（一）：基础
合并k个已排序的链表
HDFS 高可用分布式环境搭建
合并两个有序数组
连续子数组的最大和
HDFS 分布式环境搭建
容器盛水问题
大数加法
HDFS 伪分布式环境搭建
设计LRU缓存结构
两数之和
使用单调栈来解决的一些问题

zl程序教程

当前栏目

python kafka client--confluent-kafka-python

相关文章