問題描述

使用本地kafka bin/kafka-console-producer.sh --broker-list kafkaIP:port --topic topicName創(chuàng)建命令行生產(chǎn)數(shù)據(jù)，然后打開python

from pykafka import KafkaClientclient = KafkaClient(hosts='192.168.x.x:9092')topic = client.topics[’wr_test’]consumer = topic.get_balanced_consumer(consumer_group=’test-consumer-group’,auto_commit_enable=True,zookeeper_connect=’192.168.x.x:2121’)

然后自己編寫了簡單的一套處理函數(shù)，從外部引用。將數(shù)據(jù)處理后存入elasticsearch 或者 數(shù)據(jù)庫比如for msg in consumer:

if msg is not None: 外部引入的處理函數(shù)(msg.value)

在python命令行for msg in consumer:

print msg.offset, msg.value

這時候使用生產(chǎn)者敲入一些數(shù)據(jù)，在消費(fèi)端就會就會立即打印出來但是寫成py文件之后，每次運(yùn)行只會處理最近的生產(chǎn)的一次內(nèi)容，在生產(chǎn)者中再進(jìn)行輸入一些內(nèi)容，py文件就不會再進(jìn)行數(shù)據(jù)處理了。所以向問下如何編寫能運(yùn)行后能一直對消費(fèi)者數(shù)據(jù)進(jìn)行處理的函數(shù)？要注意哪些地方？

另外，get_balanced_consumer的方法，是連接zookeeper消費(fèi)使用topic.get_simple_consumer是直接消費(fèi)kafka，使用這種方式就提示No handler for...的錯誤

還有一個疑問，就是實際生產(chǎn)環(huán)境日志產(chǎn)生量很快，應(yīng)該如何編寫一個多線程處理方法？

問題解答

在別人的博客看到一種替代的解決方案http://www.cnblogs.com/castle...從consumer中將msg.value讀取到一個列表當(dāng)中，然后從列表中讀取數(shù)據(jù)進(jìn)行數(shù)據(jù)處理，當(dāng)這個流程結(jié)束后，再把列表中獲取的數(shù)據(jù)pop掉。另外也要用try: ... except :... continue