通过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 Scapy随心所欲研究TCP协议栈

1. 前言 如果只需要研究Linux的tcp协议栈行为,只需要使用packetdrill就能够满足我的所有需求。packetdrill才是让我随心所欲地撩tcp协议栈。packetdri...

用python处理图片之打开\显示\保存图像的方法

用python处理图片之打开\显示\保存图像的方法

一提到数字图像处理,可能大多数人就会想到matlab,但matlab也有自身的缺点: 1、不开源,价格贵 2、软件容量大。一般3G以上,高版本甚至达5G以上。 3、只能做研究,不易转化成...

Windows下anaconda安装第三方包的方法小结(tensorflow、gensim为例)

Windows下anaconda安装第三方包的方法小结(tensorflow、gensim为例)

 anaconda 集成了很多科学计算中所需要的包,如numpy,scipy等等,具体查看anaconda中已经预先安装配置好的包有哪些,可以通过cmd命令,输入conda l...

详解Python 多线程 Timer定时器/延迟执行、Event事件

Timer继承子Thread类,是Thread的子类,也是线程类,具有线程的能力和特征。这个类用来定义多久执行一个函数。 它的实例是能够延迟执行目标函数的线程,在真正执行目标函数之前,都...

详解python并发获取snmp信息及性能测试

python & snmp 用python获取snmp信息有多个现成的库可以使用,其中比较常用的是netsnmp和pysnmp两个库。网上有较多的关于两个库的例子。 本文重点在于如何并...