1.创建主题 

kafka-topics --create --topic topic123 --partitions 1 --replication-factor 1 --bootstrap-server 192.168.1.7:9092
[root@oracle1 bin]# kafka-topics.sh --create --topic topic123 --partitions 1 --replication-factor 1 --bootstrap-server 192.168.1.7:9092
Created topic topic123.

kafka-topics.sh --create --topic topic1 --partitions 1 --replication-factor 1 --bootstrap-server 192.168.1.7:9092
kafka-topics.sh --create --topic topic2 --partitions 1 --replication-factor 1 --bootstrap-server 192.168.1.7:9092
kafka-topics.sh --create --topic topic3 --partitions 1 --replication-factor 1 --bootstrap-server 192.168.1.7:9092
[root@oracle1 bin]# kafka-topics.sh --create --topic topic1 --partitions 1 --replication-factor 1 --bootstrap-server 192.168.1.7:9092
Created topic topic1.
[root@oracle1 bin]# kafka-topics.sh --create --topic topic2 --partitions 1 --replication-factor 1 --bootstrap-server 192.168.1.7:9092
Created topic topic2.
[root@oracle1 bin]# kafka-topics.sh --create --topic topic3 --partitions 1 --replication-factor 1 --bootstrap-server 192.168.1.7:9092
Created topic topic3.

2.查看topic 

[root@oracle1 bin]# /esdb/soft/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.1.7:2181
__consumer_offsets
sspu-nginx
topic1
topic123
topic2
topic3

3.删除topic

kafka-topics.sh --delete --zookeeper 192.168.1.7:2181  --topic topic1
[root@oracle1 bin]# kafka-topics.sh --delete --zookeeper 192.168.1.7:2181  --topic topic1
Topic topic1 is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

4.启动某个topic的生产者：topic2

kafka-console-producer.sh  --bootstrap-server 192.168.1.7:9092   --topic topic2  

[root@oracle1 bin]# kafka-console-producer.sh  --bootstrap-server 192.168.1.7:9092   --topic topic2 
>我是中国人
>我唉自己的祖国
>任何想要侵略我们的人
>都必须把它赶出中国
>我们必须做一个正义的人
>不畏强权
>不畏艰难
>努力向前
>

5.启动消费者 

--从头部开始消费
kafka-console-consumer.sh --bootstrap-server 192.168.1.7:9092 --topic  topic2 --from-beginning
[root@oracle1 ~]# kafka-console-consumer.sh --bootstrap-server 192.168.1.7:9092 --topic  topic2 --from-beginning
我是中国人
我唉自己的祖国
任何想要侵略我们的人
都必须把它赶出中国
我们必须做一个正义的人
不畏强权
不畏艰难
努力向前

6.生产者端写入

[root@oracle1 bin]# kafka-console-producer.sh  --bootstrap-server 192.168.1.7:9092   --topic topic2 
>我是中国人
>我唉自己的祖国
>任何想要侵略我们的人
>都必须把它赶出中国
>我们必须做一个正义的人
>不畏强权
>不畏艰难
>努力向前
>
>wo jue de ni bu shi ge haoren!
>souyi wo bu xiang heni zuo pengyou!
>qingni xianzai likai wo yuandian!
>


--消费者端检查 
[root@oracle1 ~]# kafka-console-consumer.sh --bootstrap-server 192.168.1.7:9092 --topic  topic2 --from-beginning
我是中国人
我唉自己的祖国
任何想要侵略我们的人
都必须把它赶出中国
我们必须做一个正义的人
不畏强权
不畏艰难
努力向前

wo jue de ni bu shi ge haoren!
souyi wo bu xiang heni zuo pengyou!
qingni xianzai likai wo yuandian!

7.查询消费组列表。

kafka-consumer-groups.sh  --bootstrap-server 192.168.1.7:9092 --list 
[root@oracle1 ~]# kafka-consumer-groups.sh  --bootstrap-server 192.168.1.7:9092 --list 
console-consumer-59913

8.查看某个消费者的消费位置。

kafka-consumer-groups  --bootstrap-server 192.168.1.7:9092 --describe --group console-consumer-59913 
[root@oracle1 ~]# kafka-consumer-groups.sh  --bootstrap-server 192.168.1.7:9092 --describe --group console-consumer-59913 

GROUP                  TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                                            HOST            CLIENT-ID
console-consumer-59913 topic2          0          -               12              -               consumer-console-consumer-59913-1-295edc35-1f57-4526-a9f0-a673310a21ab /192.168.1.7    consumer-console-consumer-59913-1

9.消费时指定消费者组的名称"汉字消费者组"

kafka-console-consumer.sh --bootstrap-server 192.168.1.7:9092 --topic  topic2 --from-beginning --group "汉字消费者组"

[root@oracle1 ~]# kafka-console-consumer.sh --bootstrap-server 192.168.1.7:9092 --topic  topic2 --from-beginning --group "汉字消费者组"
我是中国人
我唉自己的祖国
任何想要侵略我们的人
都必须把它赶出中国
我们必须做一个正义的人
不畏强权
不畏艰难
努力向前

wo jue de ni bu shi ge haoren!
souyi wo bu xiang heni zuo pengyou!
qingni xianzai likai wo yuandian!

10.查看消费者组

[root@oracle1 config]# kafka-consumer-groups.sh  --bootstrap-server 192.168.1.7:9092 --list 
汉字消费者组
console-consumer-59913

11.总结

kafka在进行消费时，我们给消费者指定具体的名称，这样方便分辨。不建议使用默认的消费者组。因为其无法进行分分辨具体的消费者。