通过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中list的交、并、差集获取方法示例

1. 获取两个list 的交集 # -*- coding=utf-8 -*- #方法一: a=[2,3,4,5] b=[2,5,8] tmp = [val for val in...

python的dataframe和matrix的互换方法

实例如下所示: #-*- encoding:utf-8 -*- import pandas as pd import numpy as np df = pd.DataFrame(np...

Python基于pygame实现的font游戏字体(附源码)

Python基于pygame实现的font游戏字体(附源码)

本文实例讲述了Python基于pygame实现的font游戏字体。分享给大家供大家参考,具体如下: 在pygame游戏开发中,一个友好的UI中,漂亮的字体是少不了的 今天就给大伙带来有关...

python tornado修改log输出方式

sed -i 's/StreamHandler()/StreamHandler(sys.__stdout__)/' /opt/python/python3/lib/python3.6/s...

python3的UnicodeDecodeError解决方法

python3的UnicodeDecodeError解决方法

爬虫部分解码异常 response.content.decode() # 默认使用 utf-8 出现解码异常 以下是设计的通用解码 通过 text 获取编码 # 通过...