kafka命令行操作及生产者消费者命令行控制台的测试

た入场券 2023-02-27 12:28 27阅读 0赞

### 一、kafka中Topic的增、删、改命令行操作 ###

1、查看当前服务器中的所有topic

[root@wuhao bin]# /opt/module/kafka_2.11-0.11.0.0/bin/kafka-topics.sh --zookeeper hadoopStudy:2181 --list

注：/opt/module/kafka\_2.11-0.11.0.0/bin/kafka-topics.sh   命令

hadoopStudy:2181 主机名：端口号

--zookeeper hadoopStudy:2181  表示所要连接的主机  
2、创建topic

[root@wuhao bin]# /opt/module/kafka_2.11-0.11.0.0/bin/kafka-topics.sh --zookeeper hadoopStudy:2181 --create --replication-factor 3 --partitions 1 --topic first

选项说明：

--topic 定义topic 名

\--replication-factor 定义副本数

\--partitions  定义分区数

3、删除Topic

[root@wuhao bin]# /opt/module/kafka_2.11-0.11.0.0/bin/kafka-topics.sh --zookeeper hadoopStudy:2181 --delete --topic first

需要server.properties中设置delete.topic.enble=true 否则只是标记删除

4、查看某个Topic详情

[root@wuhao kafka_2.11-0.11.0.0]# bin/kafka-topics.sh --zookeeper hadoopStudy:2181 --describe --topic first

5、修改分区数

[root@wuhao kafka_2.11-0.11.0.0]# bin/kafka-topics.sh --zookeeper hadoopStudy:2181 --describe --topic first

注：relication-factor的值不能超过集群中的主机数

### 二、Kafka命令行控制台的生产者消费者测试 ###

已经创建的主题Topic : four

其中主机hadoopStudy作为生产者，hadoopStudy2作为消费者，hadoopStudy3作为离线消费者

1、在主机hadoopStudy中操作

1）启动生产者

[root@wuhao kafka_2.11-0.11.0.0]# bin/kafka-console-producer.sh --topic four --broker-list hadoopStudy:9092

2）发送消息

发送消息：helloword

first\_message

![20200718184934819.png][]

2、在主机hadoopStudy2中操作

1）启动消费者（在生产者未发送消息前消费者已启动）

①使用zookeeper

[root@wuhao kafka_2.11-0.11.0.0]# bin/kafka-console-consumer.sh --topic four --zookeeper hadoopStudy:2181

②使用bootstrap-server（新版本）

[root@hadoopstudy2 kafka_2.11-0.11.0.0]# bin/kafka-console-consumer.sh --bootstrap-server hadoopStudy:9092 --topic four

注：使用zookeeper时日志文件logs显示如下，

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70][]

使用bootstrap-server 时日志文件logs显示如下（此处以主机hadoopStudy2的日志为例）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 1][]

新版本bootstrap-server 将信息保存在kafka本地（存储在Topic），由日志文件可以看出有默认的50个分区一个副本（后面分区号以一种轮询的方式分别散发到集群中每个主机的日志文件）

2）接受生产者发送的消息

![20200718185009735.png][]

3、在主机hadoopStudy3中操作

1）启动消费者（在生产者发送消息后消费者才启动）

[root@hadoopstudy3 kafka_2.11-0.11.0.0]# bin/kafka-console-consumer.sh --topic four --zookeeper hadoopStudy:2181 --from-beginning

2）判断是否能接受到生产者发出的消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 2][]

[20200718184934819.png]: /images/20230209/7beb628db4a444089fe2b3efac86d388.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70]: /images/20230209/e23ae3f684bd465689a8626d2388077e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 1]: /images/20230209/971199e6094b42b9834368572f3772e3.png
[20200718185009735.png]: /images/20230209/dab9081022b749c4ad4fa4523ff10468.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 2]: /images/20230209/2df8d93834e04d6399204bcb1e5d94dc.png