python操作kafka

#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