Kafka命令

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 创建一个名为 adb-test，分区为20, 分区副本为3 topic
kafka-topics.sh --create --topic adb-test --partitions 20 --replication-factor 3 --bootstrap-server localhost:9092

# 创建topic
kafka-topics.sh --create --topic my_topic_name --partitions 20 --replication-factor 3 --config x=y --bootstrap-server localhost:9092

# 修改 partitions 为40
kafka-topics.sh --alter --topic my_topic_name --partitions 40 --bootstrap-server localhost:9092

# 添加配置
kafka-configs.sh --alter --entity-type topics --entity-name my_topic_name --add-config x=y  --bootstrap-server localhost:9092

# 移除配置
kafka-configs.sh --alter --entity-type topics --entity-name my_topic_name --deleteConfig x  --bootstrap-server localhost:9092

1
2
3
4
5
6
7
8

# 查看一个主题的分片和同步情况
kafka-topics.sh --describe --topic adb-test --bootstrap-server localhost:9092

# 获取某个主题在zk上的偏移量
kafka-run-class kafka.tools.GetOffsetShell --broker-list "localhost:9092" --topic adb-test

# 获取上一时刻某个主题在zk上的偏移量
kafka-run-class kafka.tools.GetOffsetShell --broker-list "localhost:9092" --topic adb-test --time -1

1

kafka-topics.sh --delete --topic adb-test --bootstrap-server localhost:9092

1

kafka-console-producer --topic adb-test --broker-list localhost:9092

1

kafka-console-consumer --topic adb-test --from-beginning --bootstrap-server localhost:9092

1
2
3
4
5

# 修改kafka分区的存储量为4GB,且保存时间为120小时
kafka-configs.sh --zookeeper "zookeeper:2181" --entity-type topics --entity-name adb-test --alter --add-config retention.bytes="4294967296" log.retention.hours=120

# 修改完存储量后的查看
kafka-configs.sh --zookeeper "zookeeper:2181" --entity-type topics --entity-name adb-test --describe

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 修改 Broker ID 0 的配置（例如，日志清理线程的数量）：
kafka-configs.sh --bootstrap-server "localhost:9092" --entity-type brokers --entity-name 0 --alter --add-config log.cleaner.threads=2

# 查看brokers详情
kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 0 --describe

# 恢复到默认配置
kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 0 --alter --delete-config log.cleaner.threads

# 为集群中的所有brokers配置参数
kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-default --alter --add-config log.cleaner.threads=2

1
2

# 向kafka写入 2000000记录,每秒发送100000, 每个消息1000bytes
kafka-producer-perf-test --topic adb-test --num-records 2000000 --record-size 1000 --throughput 100000 --producer-props bootstrap.servers=localhost:9092