python通过Pykafka库来连接kafka并收发消息

原创
2016/11/15 11:31
阅读数 4W

1.安装pykafka

pip install pykafka 

2.下载安装

git clone https://github.com/Parsely/pykafka.git

然后将下载下来的pykafka文件夹下的pykafka文件(pykafka的库文件)放到/Library/Python/2.7/site-packages/路径下即可

3.假设你有至少一个卡夫卡实例在本地运行,你可以使用pykafka连接它。

consumer.py 消费者

#!/usr/bin/python
# -*- coding:utf-8 -*-
from pykafka import KafkaClient

#kafka默认端口为9092
client = KafkaClient(hosts='192.168.1.140:9092,192.168.1.141:9092,192.168.1.142:9092')#这里连接多个客户端
topic = client.topics['test_kafka_topic']

#从zookeeper消费,zookeeper的默认端口为2181
balanced_consumer = topic.get_balanced_consumer(
    consumer_group='test_kafka_group',
    auto_commit_enable=True,  # 设置为False的时候不需要添加consumer_group,直接连接topic即可取到消息
    zookeeper_connect='192.168.1.140:2181,192.168.1.141:2181,192.168.1.142:2181'#这里就是连接多个zk
)

for message in balanced_consumer:
    # print message
    if message is not None:
        print message.offset, message.value#打印接收到的消息体的偏移个数和值

producer.py 生产者

#!/usr/bin/python
# -*- coding:utf-8 -*-

from pykafka import KafkaClient
 
client = KafkaClient(hosts ="192.168.1.140:9092,192.168.1.141:9092,192.168.1.142:9092") #可接受多个client
#查看所有的topic
client.topics
print client.topics


topic = client.topics['test_kafka_topic']#选择一个topic

message ="test message test message"
#当有了topic之后呢,可以创建一个producer,来发消息,生产kafka数据,通过字符串形式,
with topic.get_sync_producer() as producer:
    producer.produce(message)
#The example above would produce to kafka synchronously - 
#the call only returns after we have confirmation that the message made it to the cluster.
#以上的例子将产生kafka同步消息,这个调用仅仅在我们已经确认消息已经发送到集群之后

#但生产环境,为了达到高吞吐量,要采用异步的方式,通过delivery_reports =True来启用队列接口;
with topic.get_sync_producer() as producer:
     producer.produce('test message',partition_key='{}'.)
producer=topic.get_producer()
producer.produce(message)
print message

 

展开阅读全文
打赏
1
1 收藏
分享
加载中
哥哥 `producer.produce('test message',partition_key='{}' ` 能解释一下 `partition_key` 的作用吗
2017/09/05 16:35
回复
举报
更多评论
打赏
1 评论
1 收藏
1
分享
返回顶部
顶部