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分区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:–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
--partitions 1
--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
--property print.key=true //是否打印key
--property print.value=true //是否打印value
--property key.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
--topic test
--partition 0
--offset 1234
--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