OpenEuler 22.03 不依赖zookeeper安装 kafka 3.3.2集群

零：规划

        本次计划安装三台OpenEuler 22.03 版本操作系统的服务器，用于搭建 kafka和flink 集群。因为从kafka 2.8 版本以后开始不依赖 zookeeper ，同时考虑到需要找一个发布时间早于  flink 1.17 的kafka 版本且应尽量稳定，综合考虑下选择了 kafka 3.3.2。

一、准备工作

        关于服务器安装与免密、防火墙设置等操作可以参考上一篇 flink集群的搭建 。kafka可以从 中文官网 下载到kafka 3.3.2 版本，需要注意的是，每个版本的 kafka 都会用 scala 2.12和2.13，如果后续的使用不涉及 scala 的部分，那么下载哪个scala版本的 kafka都无所谓。

        另外，kafka 的常用端口包括9092和9093 两个，前者用于接收集群producer和consumer等的通信，后者用于集群内部的管理和监控。

二、安装

       2.1、登录到 192.168.159.133 服务器，执行 如下命令下载安装包并解压到 /usr/local/路径下

## 进入 /usr/local/
[root@flink01 local]# cd /usr/local/

## 下载
[root@flink01 local]# wget https://archive.apache.org/dist/kafka/3.3.2/kafka_2.13-3.3.2.tgz

## 解压
[root@flink01 local]# tar -vxf kafka_2.13-3.3.2.tgz

## 进入目录、备份配置文件并修改
[root@flink01 local]# cd /usr/local/kafka_2.12-3.3.2/config/kraft

[root@flink01 local]# cp -rf server.properties server.properties_20241222bak

[root@flink01 local]# vim server.properties

        server.properties 文件需要修改的配置项如下面所示。 

## 需要关注的配置项主要包括以下几个
## 为不同kafka 服务分配不同的 id值，133服务器作为第一个节点，其他服务id随ip的递增而增加
node.id=1

## 需要确定当前 kafka 服务的通信端口和监控管理端口，这里使用的是kafka默认的 9092和9093
listeners=PLAINTEXT://:9092,CONTROLLER://:9093

## 接入集群的kafka 服务的ip 和 编号。端口号使用监控和管理端口  9093
controller.quorum.voters=1@192.168.159.133:9093,2@192.168.159.134:9093,3@192.168.159.135:9093

## 设定当前服务的接收生产者与消费者的ip和端口。
advertised.listeners=PLAINTEXT://192.168.159.133:9092

        2.2、修改 133 kafka 服务器的配置文件后，将kafka 程序目录发送到另外两个服务器

## 将kafka文件夹分发到其他服务器
[root@flink01 local]# scp kafka_2.12-3.3.2 192.168.159.134:/usr/local/

## 其他服务器上的kafka 仅需修改  server.properties  文件中的以下配置项即可，其他项保持不变

## id序号逐渐递增即可
node.id=1

## 这一项修改成当前服务器的 ip即可，端口不必改变
advertised.listeners=PLAINTEXT://192.168.159.134:9092

        2.3、创建 kafka集群 UUID，并启动集群中所有 kafka 服务

## 登入kafka 的bin目录
[root@flink03 bin]# cd /usr/local/kafka_2.12-3.3.2/bin

## 生成 集群 uuid
[root@flink01 bin]# ./kafka-storage.sh random-uuid

## 使用生成的 uuid   8xXSdiD0RdKhq-wvOpApcg  格式化集群中所有kafka服务
## 注意，要格式化集群中所有服务！
[root@flink01 bin]# ./kafka-storage.sh format -t 8xXSdiD0RdKhq-wvOpApcg -c ../config/kraft/server.properties

## 分别启动集群中三台服务器上的 kafka 服务
[root@flink01 bin]# ./kafka-server-start.sh -daemon ../config/kraft/server.properties

        2.4、创建topic 。

        事实上，在集群中任何一个 ip 的kafka下创建 topic后，其他kafka服务都能用任意一个ip 访问该名称的 topic，请自行尝试，在此不做演示。

[root@flink03 bin]#  ./kafka-topics.sh --bootstrap-server 192.168.159.135:9092 --create --topic zgyKraft --partitions 1 --replication-factor 1

        2.5、启动 consumer 和 producer，并发送消息

## 这里在第二台服务器上，启动第一台服务器的 consumer
[root@flink02 bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.159.133:9092 --topic zgyKraft

## 这里在第一台服务器上，启动第三台服务器的 producer
[root@flink01 bin]# ./kafka-console-producer.sh --broker-list 192.168.159.135:9092 --topic zgyKraft

## 然后在producer 上发送消息后，所有同名 topic 下的消费者都可以收到了。

三、附录

        附上一些常用的 kafka命令

## 查看主题
./kafka-topics.sh --bootstrap-server 192.168.159.133:9092 --list

## 查看主题明细
./kafka-topics.sh --bootstrap-server 192.168.159.133:9092 --describe <topic-id>

## 创建主题，分区 partition 为5，副本 replication-factor 为2，broker 数应 大于等于 副本数。（broker 不必在创建 topic 时显示指定）
./kafka-topics.sh --bootstrap-server 192.168.159.133:9092 --create --topic zgyTopic --partitions 5 --replication-factor 2

## 删除主题
./kafka-topics.sh --bootstrap-server 192.168.159.133:9092 --delete --topic <topic-id>

## 查看消费者列表--list
./kafka-consumer-groups.sh --bootstrap-server 192.168.159.133:9092 --list

## 查看指定消费组详情
./kafka-consumer-groups.sh --bootstrap-server 192.168.159.133:9092 --describe --group <group-id>

## 删除特定group
kafka-consumer-groups.sh --bootstrap-server 192.168.159.133:9092 --delete --group <group-id>

## 打开一个生产者
./kafka-console-producer.sh --bootstrap-server 192.168.159.133:9092 --topic zgyTopic

## 打开一个消费者
./kafka-console-consumer.sh --bootstrap-server 192.168.159.133:9092 --topic zgyTopic --from-beginning 

## 查看所有消费组详情--all-groups
./kafka-consumer-groups.sh --bootstrap-server 192.168.159.133:9092 --describe --all-groups

查询消费者成员信息--members

## 所有消费组成员信息
./kafka-consumer-groups.sh --bootstrap-server 192.168.159.133:9092 --describe --all-groups --members

## 指定消费组成员信息
./kafka-consumer-groups.sh --bootstrap-server 192.168.159.133:9092 --describe --members --group zgyConsumerGroup

## 修改到最新offset
./kafka-consumer-groups.sh --bootstrap-server 192.168.159.133:9092 --group zgyConsumerGroup --reset-offsets --topic mytopic --to-latest --execute

## 重设位移位置
./kafka-consumer-groups.sh --bootstrap-server 192.168.159.133:9092 --group zgyConsumerGroup --reset-offsets --topic mytopic --to-offset 100 --execute