kafka 集群 ZooKeeper 模式搭建

Apache Kafka是一个开源分布式事件流平台,被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用程序

Kafka 官网:Apache Kafka

关于ZooKeeper的弃用

根据 Kafka官网信息,随着Apache Kafka 3.5版本的发布,Zookeeper现已被标记为已弃用。未来计划在Apache Kafka(4.0版)的下一个主要版本中删除ZooKeeper,该版本最快将于2024年4月发布。在弃用阶段,ZooKeeper仍然支持用于Kafka集群元数据的管理,但不建议用于新的部署。新的部署方式使用 KRaft 模式,KRaft 模式部署可以看笔者的文章《kafka 集群 KRaft 模式搭建》,考虑到一些公司仍然在使用老版本的 Kafka,故笔者写这篇文章记录 Kafka 集群Zookeeper 模式搭建

官网信息截图

笔者使用3台服务器,它们的 ip 分别是 192.168.3.232、192.168.2.90、192.168.2.11

目录

1、官网下载 Kafka

2、配置 Kafka

3、启动 Kafka 集群

4、关闭 Kafka 集群

5、使用Kafka 可视化工具查看

6、测试Kafka集群


1、官网下载 Kafka

这里笔者下载最新版3.6.0

3.6.0 版本需要至少 java8 及以上版本,笔者使用的是 java8 版本

关于 linux 安装 java,没安装过的朋友可以参考《linux 系统安装 jdk》

下载完成

将 kafka分别上传到3台linux

在3台服务器上分别创建 kafka 安装目录

mkdir /usr/local/kafka

在3台服务器上分别将 kafka 安装包解压到新创建的 kafka 目录

tar -xzf kafka_2.13-3.6.0.tgz -C /usr/local/kafka

2、配置 Kafka

进入配置目录

cd /usr/local/kafka/kafka_2.13-3.6.0/config

编辑配置文件 server.properties

vi server.properties

配置 broker.id,advertised.listeners,zookeeper.connect

broker.id 每个节点的id

advertised.listeners 本机的外网访问地址

zookeeper.connect zookeeper 地址

192.168.3.232 节点配置

advertised.listeners 笔者配置为本机地址

192.168.2.90 节点

192.168.2.11 节点

笔者zookeeper 地址是 192.168.2.130:2181

zookeeper 版本是3.8.3

关于zookeeper单机安装和集群安装可以参考:《Linux环境 安装 zookeeper》《windows环境 安装 zookeeper》《linux 使用 nginx 搭建 zookeeper 集群》

3、启动 Kafka 集群

首先启动 zookeeper

然后在3台机器上依次启动 Kafka

进入 kafka 目录

cd /usr/local/kafka/kafka_2.13-3.6.0

下面2个命令皆可

bin/kafka-server-start.sh config/server.properties

bin/kafka-server-start.sh -daemon config/server.properties

4、关闭 Kafka 集群

关闭命令

bin/kafka-server-stop.sh

在 3 个节点上分别执行关闭命令

5、使用Kafka 可视化工具查看

下载地址:https://www.kafkatool.com/download.html

运行效果

6、测试Kafka集群

新建 maven 项目,添加 Kafka 依赖

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.6.0</version>
</dependency>

笔者新建 maven项目 kafka-learn

kafka-learn 项目 pom 文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.wsjzzcbq</groupId><artifactId>kafka-learn</artifactId><version>1.0-SNAPSHOT</version><dependencies><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.6.0</version></dependency></dependencies><build><plugins><plugin><artifactId>maven-compiler-plugin</artifactId><configuration><source>11</source><target>11</target></configuration></plugin></plugins></build>
</project>

新建生产者 ProducerDemo

package com.wsjzzcbq;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;/*** Demo** @author wsjz* @date 2023/11/24*/
public class ProducerDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {Properties properties = new Properties();//配置集群节点信息properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.3.232:9092,192.168.2.90:9092,192.168.2.11:9092");//配置序列化properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());Producer<String, String> producer = new KafkaProducer<>(properties);//topic 名称是demo_topicProducerRecord<String, String> producerRecord = new ProducerRecord<>("demo_topic", "明月别枝惊鹊");RecordMetadata recordMetadata = producer.send(producerRecord).get();System.out.println(recordMetadata.topic());System.out.println(recordMetadata.partition());System.out.println(recordMetadata.offset());}
}

新建消费者 ConsumerDemo

package com.wsjzzcbq;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;/*** ConsumerDemo** @author wsjz* @date 2023/11/24*/
public class ConsumerDemo {public static void main(String[] args) {Properties properties = new Properties();// 配置集群节点信息properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.3.232:9092,192.168.2.90:9092,192.168.2.11:9092");// 消费分组名properties.put(ConsumerConfig.GROUP_ID_CONFIG, "demo_group");// 序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);// 消费者订阅主题consumer.subscribe(Arrays.asList("demo_topic"));while (true) {ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String,String> record:records) {System.out.printf("收到消息:partition=%d, offset=%d, key=%s, value=%s%n",record.partition(),record.offset(),record.key(),record.value());}}}
}

运行测试

效果图

至此完

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/190623.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Christmas is Coming,一起来DIY雪花灯啦

亲爱的朋友们&#xff0c;圣诞节即将到来&#xff01;是不是很期待和家人、朋友一起欢度圣诞夜呢&#xff1f;KCC深圳联合立创开源特别为大家精心挑选了一款氛围感满满的雪花灯&#xff0c;让我们一起创造自己的圣诞礼物~ 彩色丝印PCB版的触摸无极调光雪花灯 哇&#xff0c;立创…

VL53-400激光测距传感器

一、产品简介 先由激光二极管对准目标发射激光脉冲。经目标反射后激光向各方向散射。部分散射光返回到传感器接收器&#xff0c;被光学系统接收后成像到雪崩光电二极管上。雪崩光电二极管是一种内部具有放大功能的光学传感器&#xff0c;因此它能检测极其微弱的光信号。记录并…

6.9 Windows驱动开发:内核枚举进线程ObCall回调

在笔者上一篇文章《内核枚举Registry注册表回调》中我们通过特征码定位实现了对注册表回调的枚举&#xff0c;本篇文章LyShark将教大家如何枚举系统中的ProcessObCall进程回调以及ThreadObCall线程回调&#xff0c;之所以放在一起来讲解是因为这两中回调在枚举是都需要使用通用…

XML处理相关——学习一些处理脚本(1)——(待完善)

python脚本 import osfrom Bio.Blast import NCBIXMLimport FetchUtil #https://github.com/NiallScott/FetchUtils import randomif not os.path.exists(XML):os.mkdir(XML) if not os.path.exists(dicts):os.mkdir(dicts)def bestrecipblast(org, seed, thresh5, queueNone)…

02.PostgreSQL 查询处理期间发生了什么?

PostgreSQL 查询处理期间发生了什么&#xff1f; 文中主要内容引用自PostgreSQL指南&#xff1a;内幕探索 查询处理是PostgreSQL中最为复杂的子系统。如PostgreSQL官方文档所述&#xff0c;PostgreSQL支持SQL2011标准中的大多数特性&#xff0c;查询处理子系统能够高效地处理这…

数据库相关算法题 V1

超过经理收入的员工 超过经理收入的员工显然是要将同一张表&#xff0c;作为经理和员工表连接。这里存在两种方法&#xff0c;一种是采用WHERE SELECTa.Name AS Employee FROMEmployee AS a,Employee AS b WHEREa.ManagerId b.IdAND a.Salary > b.Salary另一种是使用JOIN…

【BUG】SpringBoot项目Long类型数据返回前端精度丢失问题

问题描述 后端再给前端返回数据&#xff0c;使用Long类型的时候存在精度丢失问题。 原因分析&#xff1a; 分布式项目中广泛使用雪花算法生成ID作为数据库表的主键&#xff0c;Long类型的雪花ID有19位&#xff0c;而前端接收Long类型用的是number类型&#xff0c;但是number…

Hdoop学习笔记(HDP)-Part.08 部署Ambari集群

目录 Part.01 关于HDP Part.02 核心组件原理 Part.03 资源规划 Part.04 基础环境配置 Part.05 Yum源配置 Part.06 安装OracleJDK Part.07 安装MySQL Part.08 部署Ambari集群 Part.09 安装OpenLDAP Part.10 创建集群 Part.11 安装Kerberos Part.12 安装HDFS Part.13 安装Ranger …

有文件实体的后门无文件实体的后门rootkit后门

有文件实体后门和无文件实体后门&RootKit后门 什么是有文件的实体后门&#xff1a; 在传统的webshell当中&#xff0c;后门代码都是可以精确定位到某一个文件上去的&#xff0c;你可以rm删除它&#xff0c;可以鼠标右键操作它&#xff0c;它是有一个文件实体对象存在的。…

对系统的 Go 版本进行升级

方法一 直接升级系统的 Go 版本 注意以下操作仅适用于&#xff1a;amd64 架构的 Centos 系统。如果需要适配其他架构&#xff0c;需要自行编写代码实现。 手动执行&#xff1a; # 显示当前版本 go version # 查看环境变量 cat /etc/profile # 进入 go 的安装目录&#xff0c;…

西南科技大学(数据结构A)期末自测练习三

一、填空题&#xff08;每空1分&#xff0c;共10分&#xff09; 1、为解决计算机主机与打印机之间速度不匹配的问题&#xff0c;通常设置一个打印数据缓冲区。主机将要输出的数据依次写入缓冲区&#xff0c;打印机则依次从缓冲区中取出数据&#xff0c;则该换缓冲区的逻辑结构…

【Java Web学习笔记】 2 - CSS入门

项目代码 零、 CSS引出 CSS 教程 官方教学文档 1.在没有CSS之前&#xff0c;我们想要修改HTML元素的样式需要为每个HTML元素单独定义样式属性&#xff0c;费心费力。所以CSS就出现了。 2.使用CSS将HTML页面的内容与样式分离提高web开发的工作效率&#xff08;针对前端开发&a…

UE小计:Unreal Engine 中 Actor 数据的 JSON 序列化

Unreal Engine 中 Actor 数据的 JSON 序列化 在游戏开发过程中&#xff0c;经常需要将游戏世界中的实体&#xff08;Actors&#xff09;信息导出到其他格式以便于分析、存储或交互。在Unreal Engine中&#xff0c;我们可以通过序列化这些实体的数据为JSON格式来实现这一目的。…

一篇短文让你彻底理解什么是逻辑门电路

一、门电路概述 门电路&#xff1a;实现基本运算、复合运算的单元电路&#xff0c;如与门、与非门、或门… 注意&#xff1a;门电路中以高/低电平表示逻辑状态的1/0 正逻辑与负逻辑&#xff1a; 正逻辑&#xff1a;高电平表示1、低电平表示0 负逻辑&#xff1a;高电平表示0、低…

k8s ingress 无法找到端点

文章目录 ingress rule无法找到端点这个注解是什么意思呢&#xff1f;为何不生效呢&#xff1f;端点无法更新&#xff1f;如何确认ingressclass呢&#xff1f;修复端点无法发现的问题多个ingress controller 架构 ingress rule无法找到端点 在vnnox-cn集群创建ingress&#xf…

数据结构(三)——算法和算法分析

&#x1f600;前言 数据结构和算法是计算机科学领域中至关重要的概念。它们为解决实际问题提供了有效的方法和步骤。算法作为解决问题的方法和步骤&#xff0c;在计算机中以指令的有限序列的形式表达。本文将介绍算法的定义、描述和程序设计等方面的内容&#xff0c;帮助您深入…

【FPGA图像处理实战】- 图像基础知识

视频图像处理是FPGA主要应用方向之一&#xff0c;很多FPGA从事或准备进入这一领域&#xff0c;我们现在开始发布新的FPGA实战专栏——FPGA图像处理。 FPGA处理视频图像处理的主要优势是流水线和并行处理运算&#xff0c;特别是现在视频分辨率越来越大&#xff0c;从720p到1080…

二维A*算法

MATLAB2016b可以正常运行 function bidirectional_ASTAR clc; clear; %% 初始化界面 n 11; % field size n x n tiles 20*20的界面 %wallpercent 0.3; % this percent of field is walls 15%的界面作为阻碍物&#xff08;墙&#xff09; cmap [1 1 1; ...% 1 - whit…

RPC之GRPC:什么是GRPC、GRPC的优缺点、GRPC使用场景

简介 gRPC是一个现代的开源高性能远程过程调用(RPC)框架&#xff0c;可以在任何环境中运行。它可以高效地连接数据中心内和跨数据中心的服务&#xff0c;支持负载平衡、跟踪、运行状况检查和身份验证。它也适用于分布式计算的最后一英里&#xff0c;将设备、移动应用程序和浏览…

linux特殊权限_suid_chattr_umask

3.3 特殊权限 如果一个文件很重要&#xff0c;需要依赖特殊权限避免其被删除。 由于特殊权限会拥有一些“特权”&#xff0c;因而用户若无特殊需要&#xff0c;不应该去打开这些权限&#xff0c;避免安全方面出现严重漏洞&#xff0c;甚至摧毁系统。3个权限是对了执行文件或目…