前提条件 按照【kafka3.8.0升级文档成功搭建kafka服务】
环境：192.168.2.91 192.168.2.93
并以192.168.2.91环境kafka自带的zookeeper作为协调器。

使用基于KRaft方式进行kafka集群搭建教程

搭建kafka-ui可视化工具

1、创建kafka集群节点192.168.2.91

（1）进入kafka_2.13-3.8.0/config/
（2）修改server.properties监听端口与对外暴露端口为19092
listeners=SASL_PLAINTEXT://:19092
advertised.listeners=SASL_PLAINTEXT://192.168.2.91:19092
（3）修改server.properties存储broker信息
log.dirs=/logs/kafka-91 #修改为kafka-91另一个节点时kafka-93保证这两个目录不同即可
（4）修改server.properties的brokerId
broker.id=91
（5） 连接zookeeper 修改
zookeeper.connect=192.168.2.91:19091
（6） 创建主题

./kafka-topics.sh --create  --bootstrap-server localhost:19092 --topic ecs  --command-config ../config/kafka_topic_jass.properties

下图为代码中的kafka.properties

2、创建kafka集群节点192.168.2.93

（1）进入kafka_2.13-3.8.0_1/config/
（2）修改server.properties监听端口与对外暴露端口为29092
listeners=SASL_PLAINTEXT://:29092
advertised.listeners=SASL_PLAINTEXT://192.168.2.93:29092
（3）修改server.properties存储broker信息
log.dirs=/logs/kafka-91 # 保证这两个目录不同即可

（4）修改server.properties的brokerId
broker.id=93

（5）连接192.168.2.91的 zookeeper 修改
zookeeper.connect=192.168.2.91:19091

3、验证 方式一 直接使用命令
进入kafka_2.13-3.8.0_1/bin,启动生产者 连接29092端口，然后发送消息

进入kafka_2.13-3.8.0/bin,启动生产者 连接19092端口，发送消息

进入kafka_2.13-3.8.0_1/bin,启动消费者 接收29092端口消息，可以看到不管是19092 还时29092 发送的消息 该消费者均可接收到

4、方式二 使用kafka-ui