Kafka常用命令大全及kafka-console-consumer.sh及参数说明

kafka常用命令

一、脚本简介

1、kafka-acls.sh #配置，查看kafka集群鉴权信息
2、kafka-configs.sh #查看，修改kafka配置
3、kafka-console-consumer.sh #消费命令
4、kafka-console-producer.sh #生产命令
5、kafka-consumer-groups.sh #查看消费者组，重置消费位点等
6、kafka-consumer-perf-test.sh #kafka自带消费性能测试命令
7、kafka-mirror-maker.sh #kafka集群间同步命令
8、kafka-preferred-replica-election.sh #重新选举topic分区leader
9、kafka-producer-perf-test.sh #kafka自带生产性能测试命令
10、kafka-reassign-partitions.sh #kafka数据重平衡命令
11、kafka-run-class.sh #kafka执行脚本
12、kafka-server-start.sh #进程启动
13、kafka-server-stop.sh #进程停止
14、kafka-topics.sh #查询topic状态，新建，删除，扩容

二、常用命令大全

###1、集群相关命令###
bin/zookeeper-server-start.sh config/zookeeper.properties & # 启动zookeeper
bin/zookeeper-server-stop.sh # 停止zookeeper

bin/zkServer.sh start
bin/zkServer.sh stop

bin/kafka-server-start.sh $path/server.properties # 前台启动broker
bin/kafka-server-start.sh -daemon $path/server.properties # 后台启动broker
bin/kafka-server-stop.sh # 关闭broker

###2、topic###
bin/kafka-topics.sh --create --zookeeper $zkhost --replication-factor $replCount --partitions $partitionCount --topic $topicName #指定副本数、pritition数创建topic
bin/kafka-topics.sh --zookeeper $zkhost --alter --topic $topicName --partitions $partitionCount #扩容分区、只能扩，不能缩（涉及数据迁移、合并所以不支持））
bin/kafka-topics.sh --zookeeper $zkhost --alter --config.compression.type=gzip --topic $topicName  #动态配置topic参数
    --config.compression.type=gzip #修改或添加参数配置
    --add-config.compression.type=gzip #添加参数配置
    --delete-config.compression.type=gzip #删除参数配置
bin/kafka-topics.sh --zookeeper $zkhost --delete --topic $topicName #删除topic、需关注配置文件delete.topic.enble=true,否则无法真正删除
bin/kafka-topics.sh --zookeeper $zkhost --list #查看topic列表
bin/kafka-topics.sh --zookeeper $zkhost --describe #查看所有topic详细信息
    --topic $topicname   #指定topic查看详细信息

####################topic详细信息####################
Topic:test    PartitionCount:3    ReplicationFactor:3    Configs:
    Topic: test    Partition: 0    Leader: 0    Replicas: 0,1,2    Isr: 0,2,1
    Topic: test    Partition: 1    Leader: 1    Replicas: 1,2,0    Isr: 1,2,0
    Topic: test    Partition: 2    Leader: 2    Replicas: 2,0,1    Isr: 2,0,1
    #Topic：topic名称 
    #prititionCount：partition数量
    #replicationFactor：副本数量
    #configs：其他配置
    #Partition：prititonId
    #Leader：leader节点
    #Replicas：挂载节点
    #ISR：存活节点
######################################################


###3、消费者组###
bin/kafka-consumer-groups.sh --bootstrap-server $nodes --list #消费者列表查询(待验证)
bin/kafka-consumer-groups.sh --bootstrap-server $nodes --describe --group $groupName #显示某个消费组的消费详情(待验证)
bin/kafka-consumer-groups.sh --bootstrap-server $nodes --group $groupname --reset-offsets --all-topics --to-earliest --execute # 重设消费者组位移(待验证) 
    --to-latest --execute # 最新处    
    --to-offset 2000 --execute # 某个位置
    --to-datetime 2019-09-15T00:00:00.000 # 调整到某个时间之后的最早位移
bin/kafka-consumer-groups.sh --zookeeper $zkhost --delete --group $groupname # 删除消费者组
#####消费组的消费详情#######
TOPIC        PARTITION    CURRENT-OFFSET       LOG-END-OFFSET    LAG         CONSUMER-ID    HOST    CLIENT-ID
topic名字    分区id        当前已消费的条数    总条数        未消费的条数    消费id       主机ip    客户端id
############################


###4、生产、消费者###
bin/kafka-console-producer.sh --broker-list $nodes --topic $topicName # 使用生产者
bin/kafka-console-consumer.sh --bootstrap-server $nodes --topic $topicName --from-beginning # 使用消费者
    --from-beginning #为可选参数，表示要从头消费消息
    --from-earliest #从最早的消息开始消费（待验证）
    --from-latest #从最新的消息开始消费
    --指定offset #从指定的位置开始消费
bin/kafka-console-consumer.sh --bootstrap-server $nodes --topic $topicName --from-beginning --consumer-property group.id=$groupname # 指定groupid
bin/kafka-console-consumer.sh --bootstrap-server $nodes --topic $topicName --from-beginning --partition $partitionId # 指定分区
bin/kafka-console-consumer.sh --bootstrap-server $nodes --topic $topicName --new-consumer --from-beginning --consumer.config config/consumer.properties # 新消费者
bin/kafka-simple-consumer-shell.sh --brist localhost:9092 --topic test --partition 0 --offset 1234  --max-messages 10 # 高级点的用法(不常使用，未验证)


###5、切换leader###
./bin/kafka-preferred-replica-election.sh --zookeeper 10.161.68.123:2181,10.161.68.124:2181,10.161.68.125:2181  #切换leader，此处指定一个node


###6、压测命令###
bin/kafka-producer-perf-test.sh --topic $topicName --num-records 100 --record-size 1 --throughput 100  --producer-props bootstrap.servers=$nodes #（未验证）


###7、迁移分区###
#可先修改保存时间避免大量数据同步#
bin/kafka-configs.sh --zookeeper=$zkhost --alter --entity-type topics --add-config retention.ms=86400000 --entity-name $topicName


#创建json，json格式如下#
{
   "version":1, "partitions":[
{
   "topic":"__consumer_offsets","partition":0,"replicas":[0,1]},
{
   "topic":"__consumer_offsets","partition":1,"replicas":[0,1]},
{
   "topic":"__consumer_offsets","partition":2,"replicas":[0,1]},
{
   "topic":"__consumer_offsets","partition":3,"replicas":[0,1]},
{
   "topic":"__consumer_offsets","partition":4,"replicas":[0,1]},
{
   "topic":"__consumer_offsets","partition":5,"replicas":[0,1]},
{
   "topic":"__consumer_offsets","partition":6,"replicas":[0,1]}
]}
#执行#
bin/kafka-reassign-partitions.sh --zookeeper $zkhost --reassignment-json-file $jsonFile --execute
#验证#
bin/kafka-reassign-partitions.sh --zookeeper $zkhost --reassignment-json-file $jsonFile --verify


###8、动态参数配置###
bin/kafka-configs.sh --zookeeper=$zkhost --alter --add-config '$具体参数' --entity-type $entitytype --entity-name $entityname #只可修改支持动态配置的参数，其他的报错
    #entityType常用的包括以下几个方面，根据entityType的不同前面选择zk 或者 broker
    topics 、 users、 brokers、 concumers、 groups 等

kafka-console-consumer.sh 详解

概述

kafka-console-consumer.sh 脚本是一个简易的消费者控制台。该 shell 脚本的功能通过调用 kafka.tools 包下的 ConsoleConsumer 类，并将提供的命令行参数全部传给该类实现。

注意：Kafka 从 2.2 版本开始将 kafka-topic.sh 脚本中的 −−zookeeper 参数标注为 “过时”，推荐使用 −−bootstrap-server 参数。若读者依旧使用的是 2.1 及以下版本，请将下述的 --bootstrap-server 参数及其值手动替换为 --zookeeper zk1:2181,zk2:2181,zk:2181。一定要注意两者参数值所指向的集群地址是不同的。

消息消费

表示从 latest 位移位置开始消费该主题的所有分区消息，即仅消费正在写入的消息。

bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --topic topicName

从开始位置消费

bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --from-beginning --topic topicName

表示从指定主题中有效的起始位移位置开始消费所有分区的消息。
显示key消费

bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --property print.key=true --topic topicName

消费出的消息结果将打印出消息体的 key 和 value。

参数说明

若还需要为你的消息添加其他属性，请参考下述列表。

=