Kafka常用命令

Posted 2023-03-02 csdn_lan

1. 启动

1. zookeeper集群启动
bin/kafka-server-start.sh -daemon config/server.properties 

2. KRaft集群启动
bin/kafka-server-start.sh -daemon config/kraft/server.properties

2. 关闭

1. zookeeper集群关闭
bin/kafka-server-stop.sh config/server.properties 

2. KRaft集群关闭
bin/kafka-server-stop.sh config/kraft/server.properties

3. 主题

1. 创建主题、分区数partitions、分区的副本数replication-factor
bin/kafka-topics.sh --create --topic topic-test --zookeeper 192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181 --replication-factor 3 --partitions 3
// 2.2版本及以上使用--bootstrap-server代替--zookeeper
bin/kafka-topics.sh --create --topic topic-test --bootstrap-server 192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181 --replication-factor 3 --partitions 3

2. 列出所有主题
bin/kafka-topics.sh --bootstrap-server 192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181 --list

3. 查询主题
bin/kafka-topics.sh --describe --bootstrap-server 192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181 --topic topic-test

4. 增加主题的partition数
bin/kafka-topics.sh --bootstrap-server 192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181 --alter --topic topic-test --partitions 5 

5. 查看主题指定分区offset的最大值或最小值，time为-1时表示最大值，为-2时表示最小值：
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic topic-test --time -1 --192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181 --partitions 0

4. 生产消息

bin/kafka-console-producer.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --topic topic-test
Hello 1
Hello 2
Hello World

5. 消费消息

1. 从头开始
bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --topic topic-test --from-beginning

2. 从尾部开始
bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --topic topic-test

3. 从尾部开始,指定分区
bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --topic topic-test --offset latest --partition 0

4. 取指定个数
bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --topic topic-test --offset latest --partition 0 --max-messages 1 

5. 指定Group
bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --topic topic-test -group group-test --from-beginning

6. 消费者组

1. 查看消费者组列表
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --list

2. 查看Group详情[用来判断是否有延迟数据]
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --describe --group group-test 

3. 删除Group
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --group group-test --delete

7. 平衡Leader

1. bin/kafka-preferred-replica-election.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092
// 如果要重新分配leader的partition编号，使用--partition
bin/kafka-leader-election.sh --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --topic topic-test --partition=2 --election-type preferred

8. 使用自带压测工具压测

$ bin/kafka-producer-perf-test.sh --topic topic-test --num-records 1000000 --record-size 1 --throughput 10000 --producer-props bootstrap.servers=192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092