1. kafka-0.8.x常用命令
目录
- 1.1 启停kafkaserver
- 1.2 建立和删除topic
- 1.3 查看topic的最大最小offset
- 1.4 通过控制台命令生产和消费消息
- 1.5 查看消费者状态和消费详情
- 1.6 重置消费者offset
- 1.7 查看topic的状态和分区负载详情
- 此文档适用于kafka-0.10.1及之后的版本
- 此文档所有命令默认的路径都是kafka的home，即kafka安装目录
1.1 启停kafkaserver
命令：
//启动kafka
nohup bin/kafka-server-start.sh config/server.properties > /dev/null 2>&1 &
//关闭kafka，由于是优雅启停，在进程真的结束之前可能有一些清理工作，所以不会进程不会立刻消失，等待数秒
bin/kafka-server-stop.sh config/server.properties
//等待数秒，如果kafka进程仍无法停止，执行
kill pid
//如果仍然无法停止
kill -9 pid
启动之后，可以查看进程、日志是否正常
1.2 建立和删除topic
kafka server端需要配置delete.topic.enable=true才可以删除，否则执行删除无效，只会将topic标记为删除，不会执行真正的删除
命令：
bin/kafka-topics.sh --zookeeper ZOOKEEPER_HOST1:PORT1,ZOOKEEPER_HOST2:PORT2 --create --replication-factor REPLICA_NUM --partitions PARTITION_NUM --topic TOPIC_NAME
- ZOOKEEPER_HOST是kafka所使用的zookeeper的ip地址，PORT是zookeeper监听的端口。多个host port之间用逗号隔开
- TOPIC_NAME是要创建的topic的名称
- PARTITION_NUM是要创建的topic的分区数
- REPLICA_NUM是要创建的topic的每个分区的副本数
- zookeeper集群不需要全部列上，给出一个可用的zk地址和端口即可
例如，新建一个名为new_created_topic的topic
//新建topic，topic name为new_created_topic，zk为笔者的kafka集群使用的zk
bin/kafka-topics.sh --zookeeper 172.21.37.197:22181 --create --replication-factor 3 --partitions 16 --topic new_created_topic
//查看建立的topic的状态
bin/kafka-topics.sh --zookeeper 172.21.37.197:22181 --describe --topic new_created_topic
显示：
Topic:new_created_topic PartitionCount:16 ReplicationFactor:3 Configs:
Topic: new_created_topic Partition: 0 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
Topic: new_created_topic Partition: 1 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
Topic: new_created_topic Partition: 2 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1
Topic: new_created_topic Partition: 3 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
Topic: new_created_topic Partition: 4 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
Topic: new_created_topic Partition: 5 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2
Topic: new_created_topic Partition: 6 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
Topic: new_created_topic Partition: 7 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
Topic: new_created_topic Partition: 8 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1
Topic: new_created_topic Partition: 9 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
Topic: new_created_topic Partition: 10 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
Topic: new_created_topic Partition: 11 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2
Topic: new_created_topic Partition: 12 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
Topic: new_created_topic Partition: 13 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
Topic: new_created_topic Partition: 14 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1
Topic: new_created_topic Partition: 15 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
其中：
第一列是topic名称；第二列是partition序号；第三列是leader副本所在的kafka broker id，和kafka配置的id一致
第三列是副本分配在哪些broker上，其值是broker id列表；第四列是处于同步状态的副本所在的broker id列表
//查看topic列表，将显示server上所有的topic的列表，不显示详细信息
bin/kafka-topics.sh --zookeeper 172.21.37.197:22181 --list
//删除topic，删除后执行上一步的list，可以看到topic已经被删除；如果server没有配置允许删除，则只会标记marked for deleted
bin/kafka-topics.sh --zookeeper 172.21.37.197:22181 --delete --topic new_created_topic
1.3 查看topic的最大最小offset
查看最大最小offset可以让我们知道某个topic的数据量是多少，并且知道每个partition的earliest和latest offset。这里解释下earliest offset和latest offset：
- earliest offset：最早的offset，也就是最小的offset,在早期的kafka客户端api中，称之为smallest。假设partition 0的消息是从1127到7892，那么1127就是该partition的earliest offset
- latest offset：最新offset，也就是最大的offset,在早期的kafka客户端api中，称之为largest。。假设partition 0的消息是从1127到7892，那么7892就是该partition的latest offset
该命令在查看kafka消息的存量数量或者手工调整消费者的offset时需要用到。如果kafka消费了无效的offset，即消费的offset小于实际offset最小值或者大于实际offset最大值，将返回一个错误。
命令：
//查看最小offset
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list BORKER_HOST1:PORT1,BORKER_HSOT2:PORT2 --topic TOPIC_NAME --time -2
//查看最大offset
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list BORKER_HOST1:PORT1,BORKER_HSOT2:PORT2 --topic TOPIC_NAME --time -1
例如
//这里笔者事先建立了一个名为tttttttt_topic的topic
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 172.21.37.194:39092 --topic tttttttt_topic --time -2
tttttttt_topic:2:1
tttttttt_topic:1:2
tttttttt_topic:0:7
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 172.21.37.194:39092 --topic tttttttt_topic --time -1
tttttttt_topic:2:698
tttttttt_topic:1:630
tttttttt_topic:0:639
其中partition 2的earlist offset是1，latest offset是698，则里面有698条消息；所有的加起来，便是该topic的可用信息总量。
1.4 通过控制台命令生产和消费消息
在开发测试的过程中，有时候也在生产上，为了验证我们的topic可以正常被写入消息或者可以被正常消费，通常我们需要一个简单、直接的生产/消费工具。kafka在其bin里为我们提供了这样的脚本，可以直接在kafka server上通过命令的方式模拟生产者和消费者。
1.4.1 控制台生产者-console producer
命令：
bin/kafka-console-producer.sh --broker-list BORKER_HOST1:PORT1,BORKER_HSOT2:PORT2 --topic TOPIC_NAME
- BROKER_HOST是kafka server的ip地址，PORT是server的监听端口。多个host port之间用逗号隔开
- TOPIC_NAME是要发送消息的topic名称
- 不用将所有的kafka server的ip和port都列在后面，事实上只要有一个或者的broker的ip和port配上就可以了，客户端连接上server之后会自动获取其他节点的信息
例如，发送消息到topic为tttttttt_topic
bin/kafka-console-producer.sh --zookeeper 172.21.37.194:39092 --topic tttttttt_topic
>testmessage-111
>testmessage-222
>
1.4.2 控制台消费者-console consumer
命令：
bin/kafka-console-consumer.sh --zookeeper ZOOKEEPER_HOST1:PORT1,ZOOKEEPER_HOST2:PORT2 --topic TOPIC_NAME [--from-beginning]
- ZOOKEEPER_HOST是kafka所使用的zookeeper的ip地址，PORT是zookeeper监听的端口。多个host port之间用逗号隔开
- zookeeper集群不需要全部列上，给出一个可用的zk地址和端口即可
- TOPIC_NAME是要发送消息的topic名称
- --from-beginning是可选项，如果不加，则consumer从当前的latest offset开始消费，也就是说，执行了命令之后，需要有新的消息写入，这里才能消费到；如果加上了该参数，则从earliest offset开始消费。具体需不需要看场合，但是如果加了该参数，注意大量消息的疯狂刷屏，一般我们会结合grep或者定向到文件里
例如，从topic为tttttttt_topic的主题里消费消息
bin/kafka-console-consumer.sh --zookeeper 172.21.37.197:22181 --topic tttttttt_topic
testmessage-444
testmessage-555
1.5 查看消费者状态和消费详情
注意本节的命令仅适用于将消费者信息存放在zookeeper上的情况。kafka-0.8.x系列的版本，都是默认将consumer的状态信息写在zookeeper上。如果应用自己修改了消费者信息存储的位置，那么此章节可能不适用。
有时候我们需要关心消费者应用的状态，一般消费者应用会自己通过日志获知当前消费到了哪个topic的哪个partition的哪个offset，但当消费者出问题之后，或者出于监控的原因，我们需要知道消费者的状态和详情，那么需要借助kafka提供的相关命令。
命令：
首先我们要连接到zookeeper，在一台有zookeeper客户端的机器上执行
bin/zkCli.sh -server ZOOKEEPER_HOST1:PORT1,ZOOKEEPER_HOST2:PORT2
- ZOOKEEPER_HOST是kafka所使用的zookeeper的ip地址，PORT是zookeeper监听的端口。多个host port之间用逗号隔开
- zookeeper集群不需要全部列上，给出一个可用的zk地址和端口即可
接着，通过在zk客户端上执行命令，查看消费者信息，
get /consumers/CONSUMER_GROUP_NAME/offsets/TOPIC_NAME/PARTITION_INDEX
- CONSUMER_GROUP_NAME是消费者组的名称
- TOPIC_NAME是消费的topic的名称
- PARTITION_INDEX是要查看的partition的编号
例如，查看topic为dcs_async_redis_to_db的主题，group为consumer_buffer_20181025_2的消费者信息：
[zk: 172.21.132.66:22181,172.21.132.67:22181(CONNECTED) 0] get /consumers/consumer_buffer_20181025_2/offsets/dcs_async_redis_to_db/2
119901
cZxid = 0x40069bbb3
ctime = Thu Oct 25 15:34:43 CST 2018
mZxid = 0x500007194
mtime = Mon Oct 29 13:55:54 CST 2018
pZxid = 0x40069bbb3
cversion = 0
dataVersion = 386
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
类似的，可以更换partition编号，去查看其它partition被消费到了什么位置
1.6 重置消费者offset
注意本节的命令仅适用于将消费者信息存放在broker上的情况。对于kafka 0.10.1之后，默认的java api都是将消费信息保存在broker上，那么适用于以下命令；如果是早期版本，或者是人为将消费者信息保存在zk或者其他地方，那么此处的命令将无效。如果是保存在zk上，可以参照第二章的相关命令。
通过上一节，我们知道了如何查看消费者详情，那么在生产实践中，有时我们可能希望认为修改消费者消费到的offset位置，以达到重新消费，或者跳过一部分消息的目的，这时候重置offset的工具就非常实用。
命令：
与上一节类似，重置消费者的offset，也是仅限于基于zookeeper存储消费者信息的模式。那么第一步是登录zookeeper
bin/zkCli.sh -server ZOOKEEPER_HOST1:PORT1,ZOOKEEPER_HOST2:PORT2
- ZOOKEEPER_HOST是kafka所使用的zookeeper的ip地址，PORT是zookeeper监听的端口。多个host port之间用逗号隔开
- zookeeper集群不需要全部列上，给出一个可用的zk地址和端口即可
登录之后，执行set即可
set /consumers/CONSUMER_GROUP_NAME/offsets/TOPIC_NAME/PARTITION_INDEX NEW_OFFSET
- CONSUMER_GROUP_NAME是消费者组的名称
- TOPIC_NAME是消费的topic的名称
- PARTITION_INDEX是要查看的partition的编号
- NEW_OFFSET是想设置的offset值
注意NEW_OFFSET应该被设置成合理的offset值，不可以超过partition的earlist offset和largest offset，否则消费者可能会接收到OffsetOutofRange的异常，当收到该异常时，消费者可能会根据不同的策略选择从earliest开始继续消费或者从latest开始消费
例如，消费者将dcs_async_redis_to_db的信息全部都消费了，假设此时我想将每个分区的offset都拉到200去，从200往后重新消费，则通过命令
[zk: 172.21.132.66:22181,172.21.132.67:22181(CONNECTED) 1] set /consumers/consumer_buffer_20181025_2/offsets/dcs_async_redis_to_db/2 100
cZxid = 0x40069bbb3
ctime = Thu Oct 25 15:34:43 CST 2018
mZxid = 0x6000289a7
mtime = Wed Oct 31 21:54:58 CST 2018
pZxid = 0x40069bbb3
cversion = 0
dataVersion = 387
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0
即可，这时候可以使用上一节的命令进行检查
[zk: 172.21.132.66:22181,172.21.132.67:22181(CONNECTED) 2] get /consumers/consumer_buffer_20181025_2/offsets/dcs_async_redis_to_db/2
100
cZxid = 0x40069bbb3
ctime = Thu Oct 25 15:34:43 CST 2018
mZxid = 0x6000289a7
mtime = Wed Oct 31 21:54:58 CST 2018
pZxid = 0x40069bbb3
cversion = 0
dataVersion = 387
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0 -
可以看到offset已经被调整到了200
在生产实践中，如果需要修改消费者消费的offset，可能需要编写一个脚本来达到批量的设置，否则在分区数很多的情况下，一个个设置比较低效，且可能出错。
注意：需要先停掉消费者，才可以成功执行该命令
另，可以通过直接更换消费者group id的方式，配合消费者默认的消费策略，可以达到类似的效果，反而更加简单、高效和安全。
1.7 查看topic的状态和分区负载详情
当broker出现宕机，恢复之后，我们可以看下topic的leader是否负载均衡。因为kafka的所有读写消息的请求，都是发送到partition leader上的，因此在生产环境，负载均衡显得尤其重要。
命令：
bin/kafka-topics.sh --zookeeper ZOOKEEPER_HOST1:PORT1,ZOOKEEPER_HOST2:PORT2 --describe --topic TOPIC_NAME
- ZOOKEEPER_HOST是kafka所使用的zookeeper的ip地址，PORT是zookeeper监听的端口。多个host port之间用逗号隔开
- 类似的，zookeeper集群不需要全部列上，给出一个可用的zk地址和端口即可
例如，查看topic为dcs_storm_collect_info_ios的负载信息：
bin/kafka-topics.sh --zookeeper 172.21.22.161:2181 --describe --topic dcs_storm_collect_info_ios
显示信息 ：
Topic:dcs_storm_collect_info_ios PartitionCount:16 ReplicationFactor:2 Configs:
Topic: dcs_storm_collect_info_ios Partition: 0 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 1 Leader: 1 Replicas: 1,0 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 2 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 3 Leader: 1 Replicas: 1,0 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 4 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 5 Leader: 1 Replicas: 1,0 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 6 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 7 Leader: 1 Replicas: 1,0 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 8 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 9 Leader: 1 Replicas: 1,0 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 10 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 11 Leader: 1 Replicas: 1,0 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 12 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 13 Leader: 1 Replicas: 1,0 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 14 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: dcs_storm_collect_info_ios Partition: 15 Leader: 1 Replicas: 1,0 Isr: 0,1
本例中查看的是topic dcs_storm_collect_info_ios，其分区数是16（PartitionCount:16），副本数是2（ReplicationFactor:2）
根据第一行的分区信息
Topic: dcs_storm_collect_info_ios Partition: 0 Leader: 0 Replicas: 0,1 Isr: 0,1
我们知道编号为0的分区，其副本在broker id 0 和id 1上（Replicas: 0,1）
其分区的首领也就是leader是broker 0，也就是编号为0的那个kafka节点（Leader: 0）
在其所有副本（分布在0和1上）中，处于同步着的状态副本是0和1（Isr: 0,1）
假设此时broker 0宕机了，那么应该看到的信息会是
Topic: dcs_storm_collect_info_ios Partition: 0 Leader: 1 Replicas: 1 Isr: 1
如果发现以下现象说明kafka异常：
- 某个topic的每个分区，同步副本数量和设定的副本数量不一致
- 某个topic的每个分区，leader的id数值是-1或者none
