kafka集群操作命令「建议收藏」

全栈程序员-站长 • 2022年5月16日下午12:40 • 未分类 • 阅读 40

大家好，又见面了，我是你们的朋友全栈君。

1.修改kafka配置文件

broker.id=0
zookeeper.connect=192.168.1.10:2181,192.168.1.12:2181,192.168.1.13:2181/kafka

说明：

默认Kafka会使用ZooKeeper默认的/路径，这样有关Kafka的ZooKeeper配置就会散落在根路径下面，如果你有其他的应用也在使用ZooKeeper集群，查看ZooKeeper中数据可能会不直观，所以强烈建议指定一个chroot路径，直接在 zookeeper.connect配置项中指定。

而且，需要手动在ZooKeeper中创建路径/kafka，使用如下命令连接到任意一台 ZooKeeper服务器：

cd ~/zookeeper
bin/zkCli.sh 
create /kafka ''
在ZooKeeper执行如下命令创建chroot路径。
这样，每次连接Kafka集群的时候（使用--zookeeper选项），也必须使用带chroot路径的连接字符串，后面会看到。

2.创建topic

bin/kafka-topics.sh --create --zookeeper 192.168.1.11:2181,192.168.1.12:2181,192.168.1.13:2181/kafka --replication-factor 2 --partitions 3 --topic TEST_123

说明：

上面是创建了一个名为：test-topic的topic，分区数为3，副本数2

3.查看topic

bin/kafka-topics.sh --describe --zookeeper 192.168.1.11:2181,192.168.1.12:2181,192.168.1.13:2181/kafka

说明：

上面是查看zk集群上，当前所有topic的详细情况，如下图：

kafka集群操作命令「建议收藏」
上面Leader、Replicas、Isr的含义如下：

1 Partition：分区
2 Leader ：负责读写指定分区的节点
3 Replicas ：复制该分区log的节点列表
4 Isr ： “in-sync” replicas，当前活跃的副本列表（是一个子集），并且可能成为Leader
我们可以通过Kafka自带的bin/kafka-console-producer.sh和bin/kafka-console-consumer.sh脚本，来验证演示如果发布消息、消费消息。

4.生产消息命令

bin/kafka-console-producer.sh
--broker-list
192.168.1.12:9092,
192.168.1.13:9092,
192.168.1.14:9092
--topic
TEST_123

说明：

在上面的kafka broker上的topic上生产消息

5.消费消息命令

bin/kafka-console-consumer.sh --zookeeper 192.168.169.91:2181,192.168.1.12:2181,192.168.169.93:2181/kafka --from-beginning --topic TEST_123

说明：

可以在Producer终端上输入字符串消息行，就可以在Consumer终端上看到消费者消费的消息内容。
也可以参考Kafka的Producer和Consumer的Java API，通过API编码的方式来实现消息生产和消费的处理逻辑。

6.启动集群命令

bin/kafka-server-start.sh config/server.properties &

7.停止集群命令

bin/kafka-server-stop.sh

8.删除一个topic

bin/kafka-topics.sh  --delete --zookeeper 192.168.1.11:2181/kafka  --topic TEST_123

说明：

（1）配置文件server.properties中必须设置delete.topic.enable=true，否则只会标记为删除，而不是真正删除。

（2）执行此脚本的时候，topic的数据会同时被删除。如果由于某些原因导致topic的数据不能完全删除（如其中一个broker down了）,此时topic只会被marked for deletion，而不会真正删除。此时创建同名的topic会有冲突。

9.修改topic

（1）修改分区数量：

bin/kafka-topics.sh --alter  --zookeeper 192.168.1.11:2181/kafka --topic TEST_123 --partitions 4

（2）增加、修改或者删除一个配置参数：

bin/kafka-topics.sh —alter --zookeeper 192.168.1.11:2181/kafka  --topic TESt_123 --config key=value
bin/kafka-topics.sh —alter --zookeeper 192.168.1.11:2181/kafka  --topic
my_topic_name --deleteConfig key

发布者：全栈程序员-站长，转载请注明出处：https://javaforall.net/145869.html原文链接：https://javaforall.net