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

yipeiwu_com6年前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记录程序运行时间的三种方法

python记录程序运行时间的三种方法             ...

Python编程实现二分法和牛顿迭代法求平方根代码

Python编程实现二分法和牛顿迭代法求平方根代码

求一个数的平方根函数sqrt(int num) ,在大多数语言中都提供实现。那么要求一个数的平方根,是怎么实现的呢? 实际上求平方根的算法方法主要有两种:二分法(binary searc...

Pyqt5如何让QMessageBox按钮显示中文示例代码

Pyqt5如何让QMessageBox按钮显示中文示例代码

前言 QMessageBox是一种通用的弹出框对话框;包含:提示、警告、错误、咨询、关于等对话框;只是显示图标不同,其他功能类似; QMessageBox类常用方法如下: 按钮类型:...

手动安装python3.6的操作过程详解

手动安装python3.6只需要将其ppa源加入apt仓库列表即可,但是最近常用的一个源 ppa:jonathonf/python-3.6 突然403拒绝访问了,费劲千辛万苦终...

Python线程下使用锁的技巧分享

使用诸如Lock、RLock、Semphore之类的锁原语时,必须多加小心,锁的错误使用很容易导致死锁或相互竞争。依赖锁的代码应该保证当出现异常时可以正常的释放锁。 典型代码如下:...