当前位置： 首页 > news >正文

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

news 2024/12/24 9:54:33

零：规划

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

服务器名IP地址作用其他应用
flink01192.168.159.133kafkajdk11、flink-1.17.2
flink02192.168.159.134kafkajdk11、flink-1.17.2
flink03192.168.159.135kafkajdk11、flink-1.17.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 下的消费者都可以收到了。

三、问题与心得

        3.1、启动问题

        当我的虚拟机重启后，简单执行 2.3 环节的最后一步启动服务，发现报错，内容如下

[2024-12-24 07:39:19,194] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)
[2024-12-24 07:39:19,831] INFO Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation (org.apache.zookeeper.common.X509Util)
[2024-12-24 07:39:20,010] WARN No meta.properties file under dir /tmp/kraft-combined-logs/meta.properties (kafka.server.BrokerMetadataCheckpoint)
[2024-12-24 07:39:20,012] ERROR Exiting Kafka due to fatal exception (kafka.Kafka$)
org.apache.kafka.common.KafkaException: No `meta.properties` found in /tmp/kraft-combined-logs (have you run `kafka-storage.sh` to format the directory?)at kafka.server.BrokerMetadataCheckpoint$.$anonfun$getBrokerMetadataAndOfflineDirs$2(BrokerMetadataCheckpoint.scala:172)at scala.collection.Iterator.foreach(Iterator.scala:943)at scala.collection.Iterator.foreach$(Iterator.scala:943)at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)at scala.collection.IterableLike.foreach(IterableLike.scala:74)at scala.collection.IterableLike.foreach$(IterableLike.scala:73)at scala.collection.AbstractIterable.foreach(Iterable.scala:56)at kafka.server.BrokerMetadataCheckpoint$.getBrokerMetadataAndOfflineDirs(BrokerMetadataCheckpoint.scala:161)at kafka.server.KafkaRaftServer$.initializeLogDirs(KafkaRaftServer.scala:184)at kafka.server.KafkaRaftServer.<init>(KafkaRaftServer.scala:61)at kafka.Kafka$.buildServer(Kafka.scala:79)at kafka.Kafka$.main(Kafka.scala:87)at kafka.Kafka.main(Kafka.scala)

        经过分析，感觉需要format 操作，回顾之前的操作步骤，我只有在 2.3 环节使用 uuid时执行过 format命令。因此我生成新的集群UUID 并在三台服务器上重新执行 2.3 环节，然后问题解决。

四、附录

        附上一些常用的 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


http://www.mrgr.cn/news/81122.html

相关文章：

  • Antd react上传图片格式限制
  • 【C语言】指针数组、数组指针、函数指针、指针函数、函数指针数组、回调函数
  • 【LC】21. 合并两个有序链表
  • 访谈积鼎科技总经理：国产CFD软件发展与未来趋势展望
  • linux ipmitool配置机器的BMC（服务器管理后台）
  • Jetpack 练手项目 —— Sunflower
  • 智慧商城：编辑切换状态，删除功能，空购物车处理
  • 华为实训课笔记 2024 1223-
  • 简易CPU设计入门：本系统中的通用寄存器（一）
  • 设计模式期末复习
  • JavaScriptEs6 - String类和Array类扩展内容
  • Kamailio db_text 之使用
  • 计算机毕业设计PySpark+PyFlink+Hive地震预测系统 地震数据分析可视化 地震爬虫 大数据毕业设计 Hadoop 机器学习 深度学习
  • 基于OpenAI API使用Fastchat部署调用本地大模型
  • spring cache源码解析（四）——从@EnableCaching开始来阅读源码
  • 【数据结构练习题】栈与队列
  • 浏览器工作原理与实践-12|栈空间和堆空间：数据是如何存储的
  • 【Linux进程】进程间通信（共享内存、消息队列、信号量）
  • Jetpack 练手项目 —— Sunflower
  • 计算机毕业设计PyFlink+Hadoop广告推荐系统 广告预测 广告数据分析可视化 广告爬虫 大数据毕业设计 Spark Hive 深度学习 机器学
  • 洛谷 P2142：高精度减法 ← string+数组
  • 大语言模型驱动的Agent：定义、工作原理与应用
  • HTML5前端实现毛玻璃效果的可拖拽登录框
  • 大数据操作实验一
  • Unittest01|TestCase、断言、装饰器、夹具、清理函数、ddt
  • 计算机网络基础图解