一、topic的操作
1、查看当前服务器中的所有 topic
./kafka-topics.sh --bootstrap-server 192.168.56.93:9092 --list # 或 ./kafka-topics.sh --list --bootstrap-server 192.168.56.93:9092
2、创建 topic
./kafka-topics.sh --create --bootstrap-server 192.168.56.93:9092 --topic topicTest --partitions 1 --replication-factor 1
3、删除topic
./kafka-topics.sh --bootstrap-server 192.168.56.93:9092 --delete --topic topicTest
需要 server.properties 中设置 delete.topic.enable=true 否则只是标记删除。
4、查看topic详情
./kafka-topics.sh --bootstrap-server 192.168.56.93:9092 --describe --topic topicTest2
5、修改分区数
./kafka-topics.sh --bootstrap-server 192.168.56.93:9092 --alter --topic topicTest2 --partitions 2
注意：分区的个数只能增不能减
二、生产和消费的操作
1、发送消息
./kafka-console-producer.sh --broker-list 192.168.56.93:9092 --topic topicTest2
执行上述命令后，客户端会和kafka的broker建立一个长连接，在该连接断开之前可以持续向broker发送消息，如下图所示：
2、消费消息
# 仅能消费执行该命令之后生产者发往broker的消息 ./kafka-console-consumer.sh --bootstrap-server 192.168.56.93:9092 --topic topicTest2 # 加上--from-beginning参数后可以从头开始消费历史消息 ./kafka-console-consumer.sh --bootstrap-server 192.168.56.93:9092 --from-beginning --topic topicTest2
和生产者一样，执行消费者的上述命令后也会和broker建立一个长连接。
注意：
①上述所有命令中都有一个 --bootstrap-server 的参数，该参数须在server.properties文件中配置，即listeners的值，且命令中填写的参数值须和server.properties中填写的保持一致，不可使用localhost等替换填写的ip，否则会报Timed out waiting for a node assignmen异常
listeners=PLAINTEXT://192.168.56.93:9092
②在之前的版本中–bootstrap-server可以使用–zookeeper参数替换，填写的值也是对应的zookeeper集群的地址和端口，但新版本中已经不支持了
# 这种方式高版本的已经不支持了 ./kafka-topics.sh --zookeeper hadoop102:2181 --describe --topic topicTest2
