kafka 常用命令

1 集群操作

1.1 启动集群

1. 在每个设备启动Zookeeper：bin/zookeeper-server-start.sh config/zookeeper.properties 若不想显示日志信息：bin/zookeeper-server-start.sh config/zookeeper.properties > /dev/null 2>&1&
2. 在每个设备启动Kafka服务器：bin/kafka-server-start.sh config/server.properties 若不想显示日志信息：bin/kafka-server-start.sh config/server.properties > /dev/null 2>&1&

1.2 关闭集群

1. 在每个设备关闭Kafka服务器：bin/kafka-server-stop.sh
2. 在每个设备关闭Zookeeper：bin/zookeeper-server-stop.sh

1.3 检查Kafka状态

查看Kafka服务的状态可以使用

kafka-broker-api-versions.sh

：

bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092

1.4 重新分配分区

重新分配分区通常用于扩展Kafka集群时，将主题的分区重新分配到新的broker上。使用以下步骤进行分区重新分配：

1. 生成重新分配方案：bin/kafka-reassign-partitions.sh --generate--zookeeper localhost:2181 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2"
2. 执行重新分配：bin/kafka-reassign-partitions.sh --execute--zookeeper localhost:2181 --reassignment-json-file reassignment.json
3. 查看重新分配状态：bin/kafka-reassign-partitions.sh --verify--zookeeper localhost:2181 --reassignment-json-file reassignment.json

1.5 查看ZooKeeper中的Kafka数据

你可以使用ZooKeeper的命令行客户端

zkCli.sh

来查看Kafka在ZooKeeper中的数据：

bin/zookeeper-shell.sh localhost:2181

进入ZooKeeper shell后，可以使用以下命令：

• 查看根节点：ls /
• 查看Kafka的brokers信息：ls /brokers/ids
• 查看某个节点的数据：get /brokers/ids/0

2 主题操作

2.1 相关参数

• --topic：指定主题的名称。
• --bootstrap-server：Kafka broker的地址。
• --replication-factor：副本因子，即每个主题分区的副本数。
• --partitions：分区数。

2.2 创建主题

创建主题

test-topic

：

bin/kafka-topics.sh --create--topic test-topic --bootstrap-server localhost:9092 --replication-factor 3--partitions2

2.3 查看主题

列出所有主题：

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

描述主题

test-topic

的详细信息：

bin/kafka-topics.sh --describe--topic test-topic --bootstrap-server localhost:9092

2.4 删除主题

删除主题

test-topic

：

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

2.5 修改主题

修改主题

test-topic

：

bin/kafka-topics.sh --alter--topic test-topic --bootstrap-server your-kafka-bootstrap-server:9092 --configmax.message.bytes=10485760

3 生产与消费

3.1 生产消息

使用

kafka-console-producer.sh

发送消息到Kafka主题：

bin/kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092

启动后可以输入消息并按回车发送。

3.2 消费消息

使用

kafka-console-consumer.sh

从Kafka主题消费消息：

bin/kafka-console-consumer.sh --topic test-topic --from-beginning --bootstrap-server localhost:9092

参数说明：

• --from-beginning：表示从主题的开始位置消费消息。