devbox
寸_铁
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

部署zookeeper+kafka_kafka的版本 zk-3.8.4支持

作者：寸_铁 | 2024-08-11 11:01:09

赞

踩

kafka的版本 zk-3.8.4支持

目录

一、二进制安装

1.安装jdk

（1）yum 安装

（2）二进制安装

2.部署 zookeeper

3.部署kafka

01.安装

02.常用命令

03.示例

创建topic

创建 Kafka 生产者

创建Kafka消费者

运行示例

二、 docker 安装

1.搭建zookeeper集群

01.docker镜像拉取、docker网络创建

02.创建挂载目录

03.创建配置文件

04.zoo.cfg配置文件内容

05.集群模式启动

01.主节点启动

02.从节点1启动

03.从节点2启动

06.查看集群启动的状态

2.Kafka集群的搭建

01.拉取kafka的镜像文件

02.运行kafka_zookeeper_master节点

03.运行kafka_zookeeper_node1节点

04.运行kafka_zookeeper_node2节点

3.测试Zookeeper中Kafka节点信息

01.进入容器，开启客户端

02.查看全部节点信息

03.查看brokers

04.查看ids,kafka的broker.id以及节点下存储的详细信息

4.测试Kafka中创建Topic，Zookeeper中的接受情况

01.创建一个Topic

02.查看zookeeper集群中topic的信息

03.其它kafka节点查看这个topic

一、二进制安装

1.安装jdk

（1）yum 安装

  1. yum search java | greo jdk
  2. yum install -y java-1.8.0-openjdk
  3. java -verison

（2）二进制安装

  1. # 下载包
  2. mkdir /usr/local/java && cd /usr/local/java
  3. wget https://download.oracle.com/java/22/latest/jdk-22_linux-x64_bin.tar.gz
  4. tar -zxvf jdk-22_linux-x64_bin.tar.gz
  5. ​
  6. # 配置环境变量
  7. vi /etc/profile
  8. #set jdk
  9. JAVA_HOME=/usr/local/java/jdk-22.0.1
  10. CLASSPATH=$JAVA_HOME/lib/tools.jar
  11. PATH=$JAVA_HOME/bin:$PATH
  12. export JAVA_HOME CLASSPATH PATH
  13. ​
  14. # 使配置生效
  15. source /etc/profile
  16. ​
  17. # 查看版本
  18. java -version

2.部署 zookeeper

  1. # zookeeper 官网
  2. https://zookeeper.apache.org/
  1. # 下载安装最新稳定版 3.8.4 版本
  2. wget https://dlcdn.apache.org/zookeeper/zookeeper-3.8.4/apache-zookeeper-3.8.4-bin.tar.gz --no-check-certificate
  1. #创建工作目录
  2. mkdir /app/zookeeper
  3. tar -zxvf apache-zookeeper-3.8.4-bin.tar.gz
  1. #声明环境变量
  2. vim /etc/profile
  3. #zookeeper
  4. export ZOOKEEPER_HOME=/app/zookeeper/apache-zookeeper-3.8.4-bin
  5. export PATH=$PATH:${ZOOKEEPER_HOME}/bin
  1. # 使环境变量生效
  2. source /etc/profile
  1. #配置zookeeper
  2. cd /app/zookeeper/apache-zookeeper-3.8.4-bin/conf
  3. ​
  4. cp -p zoo_sample.cfg zoo.cfg
  5. ​
  6. vim zoo.cfg
  1. #用于计算的时间单元，以毫秒为单位，比如session超时：N*tickTime
  2. tickTime=2000
  3. #用于集群，允许从节点链接并同步到master节点的初始化连接时间，以tickTime的倍数来表示
  4. initLimit=10
  5. #用于集群，master主节点与从节点之间发送消息，请求和应答时间长度（心跳机制）
  6. syncLimit=5
  7. #连接服务器的端口，默认是2181
  8. clientPort=2181
  9. ​
  10. #存放数据文件夹
  11. dataDir=/app/zookeeper/apache-zookeeper-3.8.4-bin/data
  12. #添加日志存放文件夹
  13. dataLogDir=/app/zookeeper/apache-zookeeper-3.8.4-bin/dataLog
  14. #集群信息,地址:LF通信端口：选举端口
  15. server.1=10.88.62.179:2888:3888
  16. server.2=10.88.62.180:2888:3888
  17. server.3=10.88.62.181:2888:3888
  1. #在/app/zookeeper/apache-zookeeper-3.8.4-bin/data数据目录下创建新文件 myid ,分别写入1，2，3
  2. #主节点
  3. echo 1 > /app/zookeeper/apache-zookeeper-3.8.4-bin/data/myid
  4.  
  5. #从节点1
  6. echo 2 > /app/zookeeper/apache-zookeeper-3.8.4-bin/data/myid
  7.  
  8. #从节点2
  9. echo 3 > /app/zookeeper/apache-zookeeper-3.8.4-bin/data/myid
  1. #启动zk服务
  2. zkServer.sh start
  3. ​
  4. #查看集群状态
  5. zkServer.sh status

3.部署kafka

01.安装

  1. # 官网地址
  2. https://kafka.apache.org/downloads
  3. https://kafka.apache.org/documentation.html#gettingStarted 
  1. # 下载3.7.0版本
  2. wget https://www.apache.org/dist/kafka/3.7.0/kafka_2.12-3.7.0.tgz --no-check-certificate
  1. #创建工作目录
  2. mkdir /app/kafka
  3. tar -zxvf kafka_2.13-3.7.0.tgz
  1. #配置环境变量
  2. vim /etc/profile
  3. #kafka
  4. KAFKA_HOME=/app/kafka/kafka_2.13-3.7.0/
  5. PATH=$PATH:$KAFKA_HOME/bin/
  1. #配置文件
  2. cd /app/kafka/kafka_2.13-3.7.0/config
  3. vim server.properties
  1. #broker 的全局唯一id值，用于集群下每个 kafka 的唯一标识，可以是任意的整数值，默认为 0。
  2. #三个节点，分别修改为1,2,3
  3. broker.id=0
  4. ​
  5. #删除 topic 时是否物理删除。默认为 false 或者无此配置项(此时手动添加即可)
  6. #1、如果没有配置 delete.topic.enable，或者值为 false，则删除 topic 时是标记删除，不是真正的物理删除，在 log.dirs 配置的目录下仍然能看到数据，以及在 zk 服务器的 /brokers/topics 路径也能看到数据。
  7. #2、想要删除 topic 时真正的物理删除，此必须配置为 true.
  8. delete.topic.enable=true
  9. ​
  10. #处理网络请求与响应的线程数量，默认为 3
  11. num.network.threads=3
  12. ​
  13. #服务器用于处理请求的线程数，可能包括磁盘I/O，默认为 8
  14. num.io.threads=8
  15. ​
  16. #发送套接字的缓冲区大小，默认为 102400，100 kb
  17. socket.send.buffer.bytes=102400
  18. ​
  19. #接收套接字的缓冲区大小，默认为 102400，100 kb
  20. socket.receive.buffer.bytes=102400
  21. ​
  22. #请求套接字的缓冲区大小，默认为 104857600，100M
  23. socket.request.max.bytes=104857600
  24. ​
  25. #kafka 运行日志存放的路径，改成自定义的即可，kafka 以日志的形式存储数据，这个路径不能随意删除。
  26. log.dirs=/app/kafka/kafka_2.13-3.7.0/logs
  27. ​
  28. #topic在当前broker上的分区个数，默认为 1
  29. num.partitions=1
  30. ​
  31. #用来恢复和清理data下数据的线程数量，默认为 1
  32. num.recovery.threads.per.data.dir=1
  33. ​
  34. #日志文件保留的最长时间，超时将被删除，默认 168 小时，7 天
  35. log.retention.hours=168
  36. ​
  37. #日志文件最大大小，超过时会新建
  38. log.segment.bytes=1073741824
  39. ​
  40. #基于大小的日志（数据）保留策略，当存储的数据量超过此大小时，就删除旧数据。默认为 1G。此配置默认是被注释的。
  41. log.retention.bytes=1073741824  
  42. ​
  43. #检查日志段以查看是否可以根据保留策略删除日志段的间隔
  44. log.retention.check.interval.ms=300000
  45. ​
  46. #配置连接 Zookeeper 集群地址，默认为 localhost:2181
  47. zookeeper.connect=10.88.62.179:2181,10.88.62.180:2181,10.88.62.181:2181
  48. ​
  49. #连接到zookeeper的超时（毫秒），默认 18 秒
  50. zookeeper.connection.timeout.ms=18000
  1. #三个kafka节点，添加hosts文件，解析主机名
  2. vim /etc/hosts
  3. 10.88.62.182    iZl4y0116dv9vahi5mtdjtZ
  4. 10.88.62.183    iZl4y0116dv9vahi5mtdjuZ
  5. 10.88.62.184    iZl4y0116dv9vahi5mtdjvZ

02.常用命令

  1. #前台启动
  2. kafka-server-start.sh /app/kafka/kafka_2.13-3.7.0/config/server.properties
  1. #后台启动
  2. nohup kafka-server-start.sh -daemon /app/kafka/kafka_2.13-3.7.0/config/server.properties &
  1. #停止
  2. kafka-server-stop.sh
  1. #列出所有topic
  2. kafka-topics.sh --list --bootstrap-server 10.88.62.182:9092,10.88.62.183:9092,10.88.62.184:9092
  1. #查看topic详细信息
  2. kafka-topics.sh --bootstrap-server 10.88.62.182:9092,10.88.62.183:9092,10.88.62.184:9092 --describe --topic mytopic
  1. # 调用生产者生产消息
  2. kafka-console-producer.sh --broker-list 10.88.62.182:9092,10.88.62.183:9092,10.88.62.184:9092 --topic mytopic
  1. # 调用消费者消费消息，from-beginning表示读取全部的消息
  2. vim config/consumer.properties
  3.  
  4. # 配置 kafka服务地址，多个服务使用逗号分隔
  5. bootstrap.servers=localhost:9092
  6. # 消费者组id
  7. group.id=test-consumer-group
  1. #启动消费者
  2. kafka-console-consumer.sh --bootstrap-server 10.88.62.182:9092,10.88.62.183:9092,10.88.62.184:9092 --topic mytopic --from-beginning
  1. #消费者拉取消息
  2. kafka-console-consumer.sh --bootstrap-server 10.88.62.182:9092,10.88.62.183:9092,10.88.62.184:9092 --topic test --consumer.config ../config/consumer.properties
  1. # 删除topic
  2. kafka-topics.sh --bootstrap-server 10.88.62.182:9092,10.88.62.183:9092,10.88.62.184:9092 --delete --topic mytopic

03.示例

创建topic
kafka-topics.sh --create --topic my-topic --bootstrap-server 10.88.62.182:9092,10.88.62.183:9092,10.88.62.184:9092 --partitions 3 --replication-factor 1
创建 Kafka 生产者
  1. import org.apache.kafka.clients.producer.KafkaProducer;
  2. import org.apache.kafka.clients.producer.ProducerConfig;
  3. import org.apache.kafka.clients.producer.ProducerRecord;
  4. import org.apache.kafka.common.serialization.StringSerializer;
  5. ​
  6. import java.util.Properties;
  7. ​
  8. public class KafkaProducerExample {
  9. ​
  10.     public static void main(String[] args) {
  11.         Properties properties = new Properties();
  12.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.88.62.182:9092,10.88.62.183:9092,10.88.62.184:9092");
  13.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  14.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  15. ​
  16.         KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
  17.         String topic = "my-topic";
  18. ​
  19.         for (int i = 0; i < 10; i++) {
  20.             String message = "Message " + i;
  21.             producer.send(new ProducerRecord<>(topic, message));
  22.             System.out.println("Sent: " + message);
  23.         }
  24. ​
  25.         producer.close();
  26.     }
  27. }
创建Kafka消费者
  1. import org.apache.kafka.clients.consumer.Consumer;
  2. import org.apache.kafka.clients.consumer.ConsumerConfig;
  3. import org.apache.kafka.clients.consumer.ConsumerRecords;
  4. import org.apache.kafka.clients.consumer.KafkaConsumer;
  5. import org.apache.kafka.common.serialization.StringDeserializer;
  6. ​
  7. import java.time.Duration;
  8. import java.util.Collections;
  9. import java.util.Properties;
  10. ​
  11. public class KafkaConsumerExample {
  12. ​
  13.     public static void main(String[] args) {
  14.         Properties properties = new Properties();
  15.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.88.62.182:9092,10.88.62.183:9092,10.88.62.184:9092");
  16.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
  17.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  18.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  19. ​
  20.         Consumer<String, String> consumer = new KafkaConsumer<>(properties);
  21.         String topic = "my-topic";
  22. ​
  23.         consumer.subscribe(Collections.singletonList(topic));
  24. ​
  25.         while (true) {
  26.             ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
  27.             records.forEach(record -> {
  28.                 System.out.println("Received: " + record.value());
  29.             });
  30.         }
  31.     }
  32. }
运行示例
  1. #打开一个终端窗口，运行 Kafka 生产者示例
  2. java KafkaProducerExample
  3. ​
  4. #打开另一个终端窗口，运行 Kafka 消费者示例：
  5. java KafkaConsumerExample

二、 docker 安装

1.搭建zookeeper集群

01.docker镜像拉取、docker网络创建

  1. #Docker拉取Zookeeper的镜像文件
  2. docker pull wurstmeister/zookeeper
  3. ​
  4. #Docker创建 bridge 网络，创建自定义网络
  5. docker network create --driver bridge --subnet 172.18.0.0/16 zookeeper_network
  6. ​
  7. #查看已经存在的网络
  8. docker network ls

02.创建挂载目录

主节点创建目录

  1. # 创建 zookeeper 主节点配置存放目录
  2. mkdir -p /mydata/zookeeper_kafka/master/conf
  3. # 创建 zookeeper 主节点数据存放目录
  4. mkdir -p /mydata/zookeeper_kafka/master/data
  5. # 创建 zookeeper 主节点数据日志存放目录
  6. mkdir -p /mydata/zookeeper_kafka/master/datalog
  7. # 创建 zookeeper 主节点日志存放目录
  8. mkdir -p /mydata/zookeeper_kafka/master/logs

从节点1创建目录

  1. # 创建 zookeeper 节点1 配置存放目录
  2. mkdir -p /mydata/zookeeper_kafka/node1/conf
  3. # 创建 zookeeper 节点1 数据存放目录
  4. mkdir -p /mydata/zookeeper_kafka/node1/data
  5. # 创建 zookeeper 节点1 数据日志存放目录
  6. mkdir -p /mydata/zookeeper_kafka/node1/datalog
  7. # 创建 zookeeper 节点1 日志存放目录
  8. mkdir -p /mydata/zookeeper_kafka/node1/logs

​从节点2创建目录

  1. # 创建 zookeeper 节点2 配置存放目录
  2. mkdir -p /mydata/zookeeper_kafka/node2/conf
  3. # 创建 zookeeper 节点2 数据存放目录
  4. mkdir -p /mydata/zookeeper_kafka/node2/data
  5. # 创建 zookeeper 节点2 数据日志存放目录
  6. mkdir -p /mydata/zookeeper_kafka/node2/datalog
  7. # 创建 zookeeper 节点2 日志存放目录
  8. mkdir -p /mydata/zookeeper_kafka/node2/logs

03.创建配置文件

主节点创建配置文件

  1. # zookeeper 主节点配置存放目录
  2. cd /mydata/zookeeper_kafka/master/conf
  3. # 编辑配置文件
  4. vim zoo.cfg
  5. ​
  6. #创建myid文件
  7. echo 1 > /mydata/zookeeper_kafka/master/data/myid

从节点1配置文件

  1. # zookeeper 节点1 配置存放目录
  2. cd /mydata/zookeeper_kafka/node1/conf
  3. # 编辑配置文件
  4. vim zoo.cfg
  5. ​
  6. #创建myid文件
  7. echo 2 > /mydata/zookeeper_kafka/node1/data/myid

从节点2配置文件

  1. # zookeeper 节点2 配置存放目录
  2. cd /mydata/zookeeper_kafka/node2/conf
  3. # 编辑配置文件
  4. vim zoo.cfg
  5. ​
  6. #创建myid文件
  7. echo 3 > /mydata/zookeeper_kafka/node2/data/myid
  1. 补充:也可以直接使用cp命令，将文件复制拷贝到其他目录下:
  2.  
  3. cp zoo.cfg /mydata/zookeeper_kafka/node1/conf 
  4. cp zoo.cfg /mydata/zookeeper_kafka/node2/conf 
  5. cd /mydata/zookeeper_kafka/node2/conf

04.zoo.cfg配置文件内容

  1. # Zookeeper保存数据的目录，默认情况下，Zookeeper将写数据的日志文件也保存在这个目录里
  2. dataDir=/opt/zookeeper-3.4.13/data
  3. ​
  4. # 事物日志存储地点，如果没提供的话使用的则是 dataDir
  5. dataLogDir=/opt/zookeeper-3.4.13/datalog
  6. ​
  7. # 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳。tickTime以毫秒为单位
  8. tickTime=2000
  9. ​
  10. # 集群中的follower服务器(F)与leader服务器(L)之间初始连接时能容忍的最多心跳数（tickTime的数量）
  11. initLimit=5
  12. ​
  13. # 集群中的follower服务器与leader服务器之间请求和应答之间能容忍的最多心跳数（tickTime的数量）
  14. syncLimit=2
  15. ​
  16. # 默认值为3，不支持以系统属性方式配置。用于配置Zookeeper在自动清理的时候需要保留的快照数据文件数量和对应的事务日志文件。此参数的最小值为3，如果配置的值小于3会自动调整到3
  17. autopurge.snapRetainCount=3
  18. ​
  19. # 默认值为0，单位为小时，不支持以系统属性方式配置。用于配置Zookeeper进行历史文件自动清理的频率。如果配置为0或负数，表示不需要开启定时清理功能
  20. autopurge.purgeInterval=0
  21. ​
  22. # 默认为60，不支持以系统属性方式配置。从Socket层面限制单个客户端与单台服务器之间的并发连接数，即以ip地址来进行连接数的限制。
  23. # 如果设置为0，表示不做任何限制。仅仅是单台客户端与单个Zookeeper服务器连接数的限制，不能控制所有客户端的连接数总和
  24. maxClientCnxns=60
  25. ​
  26. # 3.5.0中的新功能：当设置为false时，可以在复制模式下启动单个服务器，单个参与者可以使用观察者运行，并且群集可以重新配置为一个节点，并且从一个节点。
  27. # 对于向后兼容性，默认值为true。可以使用QuorumPeerConfig的setStandaloneEnabled方法或通过将“standaloneEnabled = false”或“standaloneEnabled = true”添加到服务器的配置文件来设置它。
  28. standaloneEnabled=false
  29. ​
  30. # 内嵌的管理控制台，停用这个服务
  31. admin.enableServer=false
  32. ​
  33. # 开启四字命令，将所有命令添加到白名单中
  34. 4lw.commands.whitelist=*
  35. ​
  36. # 服务端口
  37. clientPort=2181
  38. ​
  39. # 集群中服务的列表
  40. server.1=172.18.0.6:2888:3888
  41. server.2=172.18.0.7:2888:3888
  42. server.3=172.18.0.8:2888:3888

05.集群模式启动

01.主节点启动
  1. # 启动命令
  2. docker run -d --restart always \
  3. --name zookeeper_kafka_master \
  4. --network zookeeper_network \
  5. --ip 172.18.0.6 \
  6. -p 2181:2181 \
  7. -e ZOO_MY_ID=1 \
  8. -v /mydata/zookeeper_kafka/master/conf/zoo.cfg:/opt/zookeeper-3.4.13/conf/zoo.cfg \
  9. -v /mydata/zookeeper_kafka/master/data:/opt/zookeeper-3.4.13/data \
  10. -v /mydata/zookeeper_kafka/master/datalog:/opt/zookeeper-3.4.13/datalog \
  11. -v /mydata/zookeeper_kafka/master/logs:/opt/zookeeper-3.4.13/logs \
  12. wurstmeister/zookeeper
02.从节点1启动
  1. # 启动命令
  2. docker run -d --restart always \
  3. --name zookeeper_kafka_node1 \
  4. --network zookeeper_network \
  5. --ip 172.18.0.7 \
  6. -p 2182:2181 \
  7. -e ZOO_MY_ID=2 \
  8. -v /mydata/zookeeper_kafka/node1/conf/zoo.cfg:/opt/zookeeper-3.4.13/conf/zoo.cfg \
  9. -v /mydata/zookeeper_kafka/node1/data:/opt/zookeeper-3.4.13/data \
  10. -v /mydata/zookeeper_kafka/node1/datalog:/opt/zookeeper-3.4.13/datalog \
  11. -v /mydata/zookeeper_kafka/node1/logs:/opt/zookeeper-3.4.13/logs \
  12. wurstmeister/zookeeper
03.从节点2启动
  1. # 启动命令
  2. docker run -d --restart always \
  3. --name zookeeper_kafka_node2 \
  4. --network zookeeper_network \
  5. --ip 172.18.0.8 \
  6. -p 2183:2181 \
  7. -e ZOO_MY_ID=3 \
  8. -v /mydata/zookeeper_kafka/node2/conf/zoo.cfg:/opt/zookeeper-3.4.13/conf/zoo.cfg \
  9. -v /mydata/zookeeper_kafka/node2/data:/opt/zookeeper-3.4.13/data \
  10. -v /mydata/zookeeper_kafka/node2/datalog:/opt/zookeeper-3.4.13/datalog \
  11. -v /mydata/zookeeper_kafka/node2/logs:/opt/zookeeper-3.4.13/logs \
  12. wurstmeister/zookeeper

06.查看集群启动的状态

  1. # 在容器 zookeeper-master 中开启一个交互模式的终端
  2. docker exec -it zookeeper_kafka_master /bin/bash
  3. ​
  4. # 查看 zookeeper 状态
  5. cd bin
  6. zkServer.sh status

2.Kafka集群的搭建

01.拉取kafka的镜像文件

docker pull wurstmeister/kafka

02.运行kafka_zookeeper_master节点

docker run -d --name kafka_zookeeper_master  --network zookeeper_network -p 9096:9096 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=172.18.0.6:2181,172.18.0.7:2181,172.18.0.8:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://10.0.4.9:9096 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9096 wurstmeister/kafka

03.运行kafka_zookeeper_node1节点

docker run -d --name kafka_zookeeper_node1 --network zookeeper_network -p 9097:9097 -e KAFKA_BROKER_ID=1 -e KAFKA_ZOOKEEPER_CONNECT=172.18.0.6:2181,172.18.0.7:2181,172.18.0.8:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://10.0.4.9:9097 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9097  wurstmeister/kafka

04.运行kafka_zookeeper_node2节点

docker run -d --name kafka_zookeeper_node2 --network zookeeper_network -p 9098:9098 -e KAFKA_BROKER_ID=2 -e KAFKA_ZOOKEEPER_CONNECT=172.18.0.6:2181,172.18.0.7:2181,172.18.0.8:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://10.0.4.9:9098 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9098  wurstmeister/kafka

3.测试Zookeeper中Kafka节点信息

01.进入容器，开启客户端

zkCli.sh

02.查看全部节点信息

  1. ls /
  2. get /controller

03.查看brokers

ls /brokers

04.查看ids,kafka的broker.id以及节点下存储的详细信息

  1. ls /brokers
  2. ls /brokers/ids
  3. ls /brokers/ids/0
  4. ls /brokers/ids/1
  5. ls /brokers/ids/2

4.测试Kafka中创建Topic，Zookeeper中的接受情况

01.创建一个Topic

kafka-topics.sh --create --topic topic_example_1 --zookeeper 172.18.0.6:2181,172.18.0.7:2181,172.18.0.8:2181 --replication-factor 3 --partitions 3

02.查看zookeeper集群中topic的信息

  1. #节点123都要查看
  2. ls /brokers
  3. ls /brokers/topics
  4. ls /brokers/topics/topic_ex
  5. ...

03.其它kafka节点查看这个topic

kafka-topics.sh --zookeeper 172.18.0.6:2181,172.18.0.7:2181,172.18.0.8:2181 --list
本文内容由网友自发贡献，转载请注明出处：https://www.wpsshop.cn/w/寸_铁/article/detail/963772
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号