生产者代码：
# -* coding:utf8 *-
from pykafka import KafkaClient
host = 'IP:9092, IP:9092, IP:9092'
client = KafkaClient(hosts = host)
print client.topics
# 生产者
topicdocu = client.topics['my-topic']
producer = topicdocu.get_producer()
for i in range(100):
print i
producer.produce('test message ' + str(i ** 2))
producer.stop()
消费者代码：
# -* coding:utf8 *-
from pykafka import KafkaClient
host = 'IP:9092, IP:9092, IP:9092'
client = KafkaClient(hosts = host)
print client.topics
# 消费者
topic = client.topics['my-topic']
consumer = topic.get_simple_consumer(consumer_group='test', auto_commit_enable=True, auto_commit_interval_ms=1,
consumer_id='test')
for message in consumer:
if message is not None:
print message.offset, message.value
以上这篇Python测试Kafka集群(pykafka)实例就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持亿速云。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
