#coding=utf-8
from pykafka import KafkaClient
import codecs
import logging
logging.basicConfig(level = logging.INFO)
client = KafkaClient(hosts = "172.16.82.163:9091")
#生产kafka数据，通过字符串形式
def produce_kafka_data(kafka_topic):
with kafka_topic.get_sync_producer() as producer:
for i in range(4):
producer.produce(‘test message‘ + str(i ** 2))
#消费kafka数据
def consume_simple_kafka(kafka_topic, timeout):
consumer = kafka_topic.get_simple_consumer(consumer_timeout_ms = timeout)
for message in consumer:
if message is not None:
print message.offset, message.value
#消费同一份kafka topic时，建议使用 get_balanced_consumer(),暂时不能使用
#问题：kazoo.handlers.threading.KazooTimeoutError: Connection time-out
def consume_kafka(kafka_topic, zkhost):
balanced_consumer = kafka_topic.get_balanced_consumer(
consumer_group = "testgroup",
auto_commit_enable = False,
zookeeper_connect = zkhost,
#zookeeper = zkhost,
zookeeper_connection_timeout_ms = 6000,
consumer_timeout_ms = 10000,
)
for message in balanced_consumer:
if message is not None:
print message.offset, message.value
#通过文件，往kafka刷数据
def produce_kafka_file(filename, kafka_topic):
with kafka_topic.get_sync_producer() as producer:
with codecs.open(filename, "r", "utf8") as rf:
for line in rf:
line = line.strip()
if not line:
continue
producer.produce(line)
#===========================================================
topic = client.topics["mytest"]
#在consumer_timeout_ms内没有任何信息返回，则中断接受消息
cosumer = topic.get_simple_consumer(consumer_timeout_ms = 10000)
cnt = 0
for message in cosumer:
if message is not None:
print message.offset, message.value
cnt += 1
print cnt