kafaka学习

创建一个topic：

[[email protected] bin]# ./kafka-topics.sh --create --zookeeper hdp1:2181  --replication-factor 1 --partitions 1 --topic justin
Created topic "justin".
[[email protected] bin]# pwd
/usr/hdp/2.2.6.0-2800/kafka/bin

在zookeeper中可以查看到刚才创建的topic:

[zk: hdp1:2181(CONNECTED) 4] ls /brokers/topics
[topic1, ambari_kafka_service_check, justin, test, kafkaToptic]

由于在创建topic的时候，指定—partitions为1，所有在zookeeper中看到的分区数为一个：

[zk: hdp1:2181(CONNECTED) 7] ls /brokers/topics/justin/partitions
[0]
[zk: hdp1:2181(CONNECTED) 8] ls /brokers/topics/justin/partitions/0
[state]
[zk: hdp1:2181(CONNECTED) 9] ls /brokers/topics/justin/partitions/0/state
[]
[zk: hdp1:2181(CONNECTED) 10] ls /brokers/topics/justin/partitions/0
[state]
[zk: hdp1:2181(CONNECTED) 11] ls /brokers/topics/justin/partitions/0/state
[]

kafka自带脚本也可以查看到zookeeper中创建的topic：

[[email protected] bin]# ./kafka-topics.sh --list --zookeeper hdp1:2181
ambari_kafka_service_check
justin
kafkaToptic
test
topic1

使用生产者，往kafka队列的justin主题（类似于bigpipe的pipe，partition和bigpipe的pipelet类似）发送两条消息：

[[email protected] bin]# ./kafka-console-producer.sh --broker-list hdp1:6667 --topic justin
justinzhang
This is another justin

使用消费者，从kafka队列的topic中获取消息，—from-beginning表示从开始处订阅：

[[email protected] bin]# ./kafka-console-consumer.sh --zookeeper hdp1:2181 --topic justin --from-beginning
justinzhang
This is another justin

可以使用将标准输入定位到文件的方式，往broker发送文件：

[[email protected] bin]# ./kafka-console-producer.sh --broker-list hdp1:6667 --topic justin < kafka-server-stop.sh

多个订阅者都可以收到生产者发送的消息。