Apache Kafka系列 命令行工具(CLI)

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Apache Kafka系列 命令行工具(CLI)相关的知识,希望对你有一定的参考价值。

Apache Kafka系列(一) 起步

Apache Kafka系列(二) 命令行工具(CLI)

Apache Kafka命令行工具(Command Line Interface,CLI),下文简称CLI。

1. 启动Kafka

  启动Kafka需要两步:

  1.1. 启动ZooKeeper 

[[email protected] kafka_2.12-0.11.0.0]# bin/zookeeper-server-start.sh config/zookeeper.properties

  1.2. 启动Kafka Server

[[email protected] kafka_2.12-0.11.0.0]# bin/kafka-server-start.sh config/server.properties 

2. 列出Topic

[[email protected] kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --zookeeper localhost:2181 --list
HelloWorld

3. 创建Topic

[[email protected] kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic Demo1
Created topic "Demo1".

  上述命令会创建一个名为Demo1的Topic,并指定了replication-factor和partitions分别为1。其中replication-factor控制一个Message会被写到多少台服务器上,因此这个值必须小于或者

  等于Broker的数量。

4. 描述Topic

[[email protected] kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic Demo1
Topic:Demo1     PartitionCount:1        ReplicationFactor:1     Configs:
        Topic: Demo1    Partition: 0    Leader: 0       Replicas: 0     Isr: 0

5. 发布消息到指定的Topic

[[email protected] kafka_2.12-0.11.0.0]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Demo1
>this
>is 
>the 
>firest
>input

  可以在控制台逐行输入任意消息。命令的终止符是:control + C组合键。

6. 消费指定Topic上的消息

[[email protected] kafka_2.12-0.11.0.0]# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic Demo1
this
is 
the 
firest
input

7. 修改Topic

  7.1 增加指定Topic的partition,在第3步中创建的Demo1的partition是1。如下命令将增加10个partition

[[email protected] kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --alter --zookeeper localhost:2181 --partitions 11 --topic Demo1
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!

  7.2. 删除指定Topic

[[email protected] kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic Demo1
Topic Demo1 is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

    Note中指出该Topic并没有真正的删除,如果真删除,需要把server.properties中的delete.topic.enable置为true

  7.3 给指定的Topic增加配置项,如给一个增加max message size值为128000

[[email protected] kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --alter --zookeeper localhost:2181 --topic Demo1 --config max.message.bytes=128000
WARNING: Altering topic configuration from this script has been deprecated and may be removed in future releases.
         Going forward, please use kafka-configs.sh for this functionality
Updated config for topic "Demo1".

    warning中指出该命令已经过期,将来可能被删除,替代的命令是使用kafka-config.sh。新命令如下:

[[email protected] kafka_2.12-0.11.0.0]# bin/kafka-configs.sh --alter --zookeeper localhost:2181 --entity-type topics --entity-name Demo1 --add-config max.message.bytes=12800
Completed Updating config for entity: topic Demo1.

    需要使用entity-type置为topics,并在entity-name中指定对应的名称

8. 结论

  本文展示了CLI所提供的一些常用的命令,这些基本的命令在运维Kafka过程中很实用。

 

以上是关于Apache Kafka系列 命令行工具(CLI)的主要内容,如果未能解决你的问题,请参考以下文章

Apache Kafka系列 Kafka Connect及FileConnector示例

Apache Kafka系列 多线程Consumer方案

Apache Commons CLI 开发命令行工具示例

Apache Common CLI 如何实现命令行解析

Apache Common CLI 如何实现命令行解析

Jenkins基础篇 系列之-—10 执行 Jenkins CLI 命令