Kafka基本操作
启动Kafka
bin/kafka-server-start.sh config/server.properties&
创建一个Topic
bin/kafka-topics.sh --create --topic test1 --partitions 1 --replication-factor 1 --zookeeper 127.0.0.1:2181
然后查看一下
bin/kafka-topics.sh --list --zookeeper 127.0.0.1:2181
创建发送者
bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic test1
创建消费者
bin/kafka-console-consumer.sh --zookeeper 127.0.0.1:2181 --from-beginning --topic test1
写一个简单的python脚本
#/usr/bin/python
#coding=utf-8
from pykafka import KafkaClient
client = KafkaClient(hosts="127.0.0.1:9092")
topics = client.topics
topic = topics['test1']
producer = topic.get_producer()
producer.produce('test message 45')
print "==================="
consumer = topic.get_simple_consumer()
for message in consumer:
if message is not None:
print message.offset, message.value