通过pykafka接收Kafka消息队列的方法

yipeiwu_com5年前Python基础

没有Kafka环境,所以也没有进行验证。感觉今后应该能用到,所以借抄在此,备查。

pykafka使用示例,自动消费最新消息,不重复消费:

# -* coding:utf8 *-
from pykafka import KafkaClient
 
host = '192.168.200.38'
client = KafkaClient(hosts="%s:9092" % host)
 
print client.topics
 
# 生产者
# topicdocu = client.topics['task_pull']
# producer = topicdocu.get_producer()
# for i in range(4):
#   print i
#   producer.produce('test message ' + str(i ** 2))
# producer.stop()
 
# 消费者
topic = client.topics['task_push']
consumer = topic.get_simple_consumer(consumer_group='test', auto_commit_enable=True, consumer_id='test')
for message in consumer:
  if message is not None:
    print message.offset, message.value

以上这篇通过pykafka接收Kafka消息队列的方法就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持【听图阁-专注于Python设计】。

相关文章

Python语言实现将图片转化为html页面

Python语言实现将图片转化为html页面

PIL 图像处理库 PIL(Python Imaging Library) 是 Python 平台的图像处理标准库。不过 PIL 暂不支持 Python3,可以用 Pillow 代替,...

Python 中 function(#) (X)格式 和 (#)在Python3.*中的注意事项

python 的语法定义和C++、matlab、java 还是很有区别的。 1. 括号与函数调用 def devided_3(x): return x/3. print(a)...

Python 单例设计模式用法实例分析

本文实例讲述了Python 单例设计模式用法。分享给大家供大家参考,具体如下: demo.py(单例): class MusicPlayer(object): # 类属性 记录对...

浅谈django model postgres的json字段编码问题

django model的json字段的编码器不能有效编码诸如uuid,datetime等数据类型,当直接存储此类型的对象到json字段中为抛出编码异常,这时可以通过JSONField字...

python的random模块及加权随机算法的python实现方法

random是用于生成随机数的,我们可以利用它随机生成数字或者选择字符串。 •random.seed(x)改变随机数生成器的种子seed。 一般不必特别去设定seed,Pyt...