之前我推荐了多个Kafka manager(or UI)，web图形化的Kafka管理工具是提升效率的利器。不过命令行的管理操作仍然是必知必会的。

对于Kafka的操作命令，我进行了如下的总结：

前提条件：如下的操作都需要在Kafka安装部署目录内操作（或者绝对路径也阔以）。

集群、服务器运维类型

前台启动：

bin/kafka-server-start.sh ../config/server.properties

-daemon 以后台的方式启动：

bin/kafka-server-start.sh -daemon ../config/server.properties

指定JMX port端口启动：

指定jmx用来监控kafka集群，jmx的port可以灵活配置，我配置的是9999。

JMX_PORT=9999 nohup ./kafka-server-start.sh ../config/server.properties > kfk_service.log  2>&1 &

or

JMX_PORT=9999 ./kafka-server-start.sh -daemon ../config/server.properties

查询读取操作类型

topic相关

查看帮助：

bin/kafka-topics.sh --help

查看当前集群的topic列表：

bin/kafka-topics.sh --bootstrap-server {您的kafka broker:port}  --list

查看指定的topic详细信息：

bin/kafka-topics.sh --bootstrap-server {您的kafka broker:port} -describe -topic {您的topic名称}

查看集群所有topic的详细信息：

bin/kafka-topics.sh --bootstrap-server {您的kafka broker:port} --describe

查看指定topic各个分区消息的数据量：

bin/kafka-run-class.sh  kafka.tools.GetOffsetShell --broker-list {您的kafka broker:port list} --topic {您的topic名称} --time -1

查看指定topic的log日志消息内容：

我这个topic的log存储路径是“/data/kafka24-logs000/canal_test-0/”

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files /data/kafka24-logs000/canal_test-0/00000000000000473378.log --print-data-log

查看指定topic某个消费组对应的offset：

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --topic {您的topic名称} --bootstrap-server {您的kafka broker:port} --group {您的消费组名称}

查看指定topic最旧offset（earliest）：

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list {您的kafka broker:port list} --topic {您的topic名称} --time -2

查看指定topic最新offset（latest）：

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list {您的kafka broker:port list} --topic {您的topic名称} --time -1

消费者相关

查看topic某个消费组对应的offset：

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --topic {您的topic名称} --bootstrap-server {您的kafka broker:port} --group {您的消费组名称}

查看集群所有消费者群组：

bin/kafka-consumer-groups.sh --bootstrap-server {您的kafka broker:port} --list

查询某个消费组的消费详细信息：

bin/kafka-consumer-groups.sh --bootstrap-server {您的kafka broker:port} --describe --group {您的消费组名称}

写操作类型

topic相关

新建指定topic：

这里带了若干参数的指定，可以不带参数

bin/kafka-topics.sh --bootstrap-server {您的kafka broker:port} --create --topic {您的topic名称} --partitions 1 --replication-factor 1 --config max.message.bytes=64000 --config flush.messages=1

修改topic配置：添加分区数为3个（注意只能+，不能-）：

bin/kafka-topics.sh --bootstrap-server {您的kafka broker:port} --alter --topic {您的topic名称} --partitions 3

删除指定topic：

bin/kafka-topics.sh  --bootstrap-server {您的kafka broker:port} --delete --topic {您的topic名称}

生产者相关

生产消息（用来测试）：

bin/kafka-console-producer.sh --broker-list {您的kafka broker:port list} --topic {您的topic名称}

消费者相关

删除消费者群组：

bin/kafka-consumer-groups.sh --bootstrap-server {您的kafka broker:port} --delete --group {您的消费组名称}

删除消费者群组中的topic：

bin/kafka-consumer-groups.sh --bootstrap-server {您的kafka broker:port} --delete --group {您的消费组名称} --topic {您的topic名称}

小结

知晓Kafka的版本是重要的，版本不同，命令操作也有差异、淘汰、升级。另外即使同一个版本的bin下的不同命令对于一些参数也没有做到统一，如果报错多查看帮助。