KAFKA-03-kafka 脚本命令使用详解

0：脚本总结

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分区leader9、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：–broker-list,–bootstrap-server以及–zookeeper 区别

• kafka-topic.sh 和 kafka-console-consumer.sh 都是使用 --bootstrap-server
• kafka-console-producer.sh 使用 --broker-list，本来这里也可以改成使用 bootstrap-server ，但是创建者没改，可能是为了兼容，也可能是忘了；
• Kafka 从 2.2 版本开始将 kafka-topic.sh 脚本中的 −−zookeeper 参数标注为 “过时”，推荐使用 −−bootstrap-server 参数。
• 若读者依旧使用的是 2.1 及以下版本，请将下述的 --bootstrap-server 参数及其值手动替换为 --zookeeper zk1:2181,zk2:2181,zk:2181。一定要注意两者参数值所指向的集群地址是不同的。
• bootstrap.servers 参数，只在启动客户端连接 Kafka 服务时使用。并且即使列表里面填了多个节点，只要有一个可用就行了。

2：主题(topic)命令行操作

1：主题命令行参数 ./bin/kafka-topics.sh

./bin/kafka-topics.sh

• –bootstrap-server <String: server toconnect to> 连接的 Kafka Broker 主机名称和端口号。
• –topic <String: topic> 操作的 topic 名称。
• –create 创建主题。
• –delete 删除主题。
• –alter 修改主题。
• –list 查看所有主题。
• –describe 查看主题详细描述。
• –partitions <Integer: # of partitions> 设置分区数。
• –replication-factor<Integer: replication factor> 设置分区副本。
• –config <String: name=value> 更新系统默认的配置。

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

2：创建topic

现在我们来创建一个名字为“topic01”的Topic，这个topic只有一个partition，并且备份因子也设置为1：

./bin/kafka-topics.sh  
        --bootstrap-server centos:9092 
        --create--partitions1 
        --replication-factor 1--topic topic01**

3：查看topic列表

./bin/kafka-topics.sh  
        --bootstrap-server centos:9092 
        --list

4：查看topic详情

./bin/kafka-topics.sh --bootstrap-server centos:9092 --describe--topic topic01

Topic:topic01    PartitionCount:3    ReplicationFactor:3    Configs:segment.bytes=1073741824
    Topic: topic01    Partition: 0    Leader: 0    Replicas: 0,2,3    Isr: 0,2,3
    Topic: topic01    Partition: 1    Leader: 2    Replicas: 2,3,0    Isr: 2,3,0
    Topic: topic01    Partition: 2    Leader: 0    Replicas: 3,0,2    Isr: 0,2,3

5：修改topic—只能增加分区和副本，不能减少

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh 
                    --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092 
                    --alter 
                    --topic topic03 
                    --partitions 2

6：删除topic

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh 
                    --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092 
                    --delete 
                    --topic topic03

2：消费者命令 - ./bin/kafka-console-consumer.sh

1：命令参数

./bin/kafka-console-consumer.sh

• –bootstrap-server <String: server toconnect to> 连接的 Kafka Broker 主机名称和端口号。
• –topic <String: topic> 操作的 topic 名称。
• –from-beginning #为可选参数，表示要从头消费消息 –from-earliest #从最早的消息开始消费 –from-latest #从最新的消息开始消费
• –group <String: consumer group id> 指定消费者组名称。
• –指定offset #从指定的位置开始消费

2：消费者订阅topic01

./bin/kafka-console-consumer.sh  
        --bootstrap-server centos:9092 
        --topic topic01
        --group g1 
        --propertyprint.key=true      //是否打印key
        --propertyprint.value=true   //是否打印value
        --propertykey.separator=,    //key和value之间的分割

3：指定groupid

# 指定groupid
./bin/kafka-console-consumer.sh 
        --bootstrap-server $nodes--topic$topicName 
        --from-beginning 
        --consumer-property group.id=$groupname

4：指定分区

# 指定分区
./bin/kafka-console-consumer.sh 
        --bootstrap-server $nodes--topic$topicName 
        --from-beginning 
        --partition$partitionId

5：新消费者

# 新消费者
./bin/kafka-console-consumer.sh 
        --bootstrap-server $nodes--topic$topicName 
        --new-consumer 
        --from-beginning 
        --consumer.config config/consumer.properties 

6：高级点的用法

# 高级点的用法
./bin/kafka-simple-consumer-shell.sh 
        --brist localhost:9092 
        --topictest--partition0--offset1234  
        --max-messages 10

3：生产者命令 - ./bin/kafka-console-producer.sh

1：生产者命令行参数

 ./bin/kafka-console-producer.sh

• –broker-list：必传参数，用来指定Kafka的代理地址列表，多个地址间使用逗号隔开
• –topic <String: topic> 操作的 topic 名称。

2：生产者生产消息

./bin/kafka-console-producer.sh 
        --broker-list centos:9092 
        --topic topic01

4：消费者组 - bin/kafka-consumer-groups.sh

#消费者列表查询
./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