2019-10-27
Kafka基本操作
1. 查看Topic
--list命令(列举所有主题)：
kafka-topics.sh –list –zookeeper server-1:2181,server-2:2181
--describe命令(不指定topic参数则查看所有主题信息，若指定topic参数则查看特定主题的信息)：
kafka-topics.sh –describe –zookeeper server-1:2181,server-2:2181
#查看指定主题已覆盖的配置:
Kafka-topics.sh –describe –zookeeper server-1:2181,server-2:2181,server-3:2181–topics-with-overrides–topic config-test
2.创建Topic
命令：
kafka-topics.sh –create–zookeeper server-1:2181,server-2:2181,server-3:2181 --replication-factor 2 –partitions3 –topic kafka-action
此时会在${log.dir}目录下创建相应的分区文件目录，副本分别分布在不同的节点上。
同时登陆ZooKeeper客户端查看所创建的主题元数据信息，登陆ZK之后，“kafka-action”元数据信息如下：
Ls /brokers/topics/kafka-action/partitions
[0,1,2]
ls /brokers/topics/kafka-action
{“version”:1,”partitions”:{“2”:[3,1], “1”:[2,3],”0”:[1,2]}}
注意:
1）创建topic时，主要要指定副本数、分区数等，如果不指定副本和分区数，默认分别创建${num.partitions}个分区数和${default.replcation.factor}个副本。
2）当生产者向一个还未创建的topic发送消息时，会自动创建该副本，但是前提是：auto.create.topics.enable=true
3. 修改Topic
当创建一个主题之后，可以通过alter命令对主题进行修改，包括修改主题级别的配置、增加主题分区、修改副本分配方案、修改主题Offset等。
命令：
kafka-topics.sh --alter –zookeeper server-1:2181,server-2:2181,server-3:2181–topic config-test –config max.message.bytes=204800
修改max.message.bytes参数值，某些版本使用kafka-topics.sh脚本修改参数可以成功，但是可能会提示该脚本已过时，在后续版本可能会被移除，推荐使用kafka-configs.sh脚本修改参数。
kafka-topics.sh --alter –zookeeper server-1:2181,server-2:2181,server-3:2181–topic config-test –configsegment.bytes=209715200
修改segment.bytes参数后，可通过topics-with-overrides查看主题config-test已覆盖的配置信息。
删除参数segment.bytes配置：
kafka-topics.sh --alter –zookeeper server-1:2181,server-2:2181,server-3:2181–topic config-test –delete-configsegment.bytes
再次通过topics-with-overrides查看主题config-test已覆盖的配置信息，此时已不包含segment.bytes。
增加分区：
Kafka并不支持减少分区的操作，只能为一个主题增加分区。
假设当前分区为3个，要增加至5个：
kafka-topics.sh --alter –zookeeper server-1:2181,server-2:2181,server-3:2181–topic config-test –partitions 5
4.删除Topic
删除Kafka主题，一般有以下两种方式：
[if !supportLists]（1） [endif]手动删除各节点${log.dir}目录下该主题分区文件夹，同时登录ZK客户端删除待删除主题对应的节点，主题元数据保存在/brokers/topics和/config/topics目录下。
[if !supportLists]（2） [endif]执行kafka-topics.sh脚本进行删除，若希望通过该脚本彻底删除主题，则需要保证在启动Kafka时所加载的server.properties文件中配置delete.topic.enable=true,该配置默认为false。否则执行该脚本并未真正删除主题，而是在ZK的/admin/delete_topics目录下创建一个与待删除主题同名的节点，将该主题标记为删除状态。
5.启动生产者发送消息
启动一个向主题kafka-action发送消息的生产者，同时指定每条消息包含有Key:
Kafka-console-producer.sh –broker-listserver-1:9092,server-2:9092,server-3:9092 –topic kafka-action –property parse.key=true
该命令执行后，控制台等待客户端输入消息。由于没有指定消息Key与消息净荷(payload)之间的分隔符，默认是以制表符分隔。可以通过配置项key.separator指定。
Kafka-console-producer.sh –broker-list server-1:9092,server-2:9092,server-3:9092–topic kafka-action –property parse.key=true –property key.separator=’ ’
在控制台分别输入一批消息，消息key与消息实际数据之间以空格分隔。然后执行以下命令，验证消息是否发送成功。
Kafka-run-class.shkafka.tools.GetOffsetShell –broker-list server-1:9092,server-2:9092,server-3:9092–topic kafka-action –time -1
以上命令用于查看某个主题各分区对应消息偏移量。可以通过partitions参数指定一个或多个分区，多个分区之间以逗号分隔，若不指定则默认查看该主题所有分区；time参数表示查看在指定时间之前的数据，支持-1(latest)、-2(earliest)两个时间选项，默认取值为-1.
执行以上命令输出结果如下，共3列，分别表示主题名、分区编号、消息偏移量：
Kafka-action:2:6
Kafka-action:1:6
Kafka-action:0:4
通过结果信息可知，总共产生了16条消息，3个分区按编号从大到小依次有6条、6条、4条消息。
6.查看消息
Kafka生产的消息以二进制的形式在文件中，为了便于查看消息内容，Kafka提供了一个查看日志文件的工具类kafka.tools.DumpLogSegments.命令如下：
Kafka-run-class.shkafka.tools.DumpLogSegments –files ./././00000000000000.log
7.启动消费者接受消息
