kafka集群搭建-使用zookeeper

1.环境准备：

使用如下3台主机搭建zookeeper集群，由于默认的9092客户端连接端口不在本次使用的云服务器开放端口范围内，故端口改为了8093。
172.2.1.69:8093
172.2.1.70:8093
172.2.1.71:8093

2.下载地址

去官网下载，或者使用如下仓库地址下载，本次使用的时kafka_2.13-3.6.1.tgz ，即3.6.1版本，前面的2.13是scala版本，该版本是较新的版本，可以使用zookeeper，也可以不使用zookeeper搭建集群，本次记录使用了zk，zk集群的部署可以参考上一篇记录。

# 软件包下载地址，可以切到/kafka/路径，选择自己需要的版本
https://archive.apache.org/dist/kafka/3.6.1/kafka_2.13-3.6.1.tgz

3.软件包下载解压

在上面3台服务器上分别执行wget下载，或者本地下载后上传，本次使用的环境为堡垒机接入，如果使用的是宿主机账密登陆，可以下载配置一台，其余使用SCP命令拷贝过去即可。

cd  /usr/local/wget https://archive.apache.org/dist/kafka/3.6.1/kafka_2.13-3.6.1.tgztar -zxvf kafka_2.13-3.6.1.tgzmv kafka_2.13-3.6.1 kafka

4.修改配置

需要修改logs路径的话，可以在/kafka路径下新建logs路径，并配置到server.properties中，这里使用默认的/tmp/kafka-logs路径。

cd kafka
vim conf/server.properties

修改内容如下，

# 每个节点唯一的id，这里.69、.70、.71服务器分别设置为了1、2、3
broker.id=1# 默认为9092，云服务器开放端口问题，改为了8093
port=8093# 上一篇记录博客搭建的zk集群地址
zookeeper.connect=172.2.1.69:8092,172.2.1.70:8092,172.2.1.71:8092# 配置监听访问、绑定地址，这里都是PLAINTEXT协议，不需要认证（相当于内网访问）
listeners=PLAINTEXT://0.0.0.0:8093                                                                                                 
advertised.listeners=PLAINTEXT://172.2.1.71:8093# 日志路径
log.dirs=/tmp/kafka-logs

5.启动kafka集群

分别在每个节点的bin路径下执行启动脚本

# 在3个节点分别执行如下命令，-daemon表示后台启动，不带该参数前台启动
./bin/kafka-server-start.sh -daemon config/server.properties

使用jps命令，或者去kafka启动日志，查看kafka是否启动成功。

6.创建topic`

# 创建topic，在任一节点执行都可以。./bin/kafka-topics.sh --bootstrap-server=172.2.1.69:8093,172.2.1.70:8093,172.2.1.71:8093 --create --topic topic-demo --partitions=3 --replication-factor=3
# 查看topic是否创建成功，在任一节点执行
./bin/kafka-topics.sh --bootstrap-server=172.2.1.69:8093,172.2.1.70:8093,172.2.1.71:8093 --topic topic-demo --list

7.模拟生产消费消息

需要注意的是网上搜到的一些老的博客kafka命令在高版本中是不再支持的，如：
sh ./bin/kafka-topics.sh --zookeeper=zk集群地址，可能出现命令无法识别：zookeeper is not a recognized option
需要替换为：sh ./bin/kafka-topics.sh --bootstrap-server=kafka集群地址，注意–bootstrap-server后面跟的是kafka集群地址，不是zookeeper地址。

# 在2个节点启动消费者模拟客户端接收消息，在第3个节点启动生产者模拟客户端发送消息
./bin/kafka-console-consumer.sh --bootstrap-server=172.2.1.69:8093,172.2.1.70:8093,172.2.1.71:8093 --topic topic-demo

# 在第3个节点启动生产者客户端模拟发送消息：hello
./bin/kafka-console-producer.sh --bootstrap-server=172.2.1.69:8093,172.2.1.70:8093,172.2.1.71:8093 --topic topic-demo

生产者客户端发送hello
在这里插入图片描述
此时可以看到2个消费者模拟客户端都受到了消息：hello
在这里插入图片描述

8.集成springboot

坐标如下：

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>

kafka配置类：

package com.example.kafka.kafka;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;import java.util.HashMap;
import java.util.Map;@Configuration
@EnableKafka
//@RefreshScope
public class KafkaConfig {@Value("${xxxx:172.2.1.69:8093,172.2.1.70:8093,172.2.1.71:8093}")private String kafkaServers;public Map<String, Object> producerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers);props.put(ProducerConfig.RETRIES_CONFIG, 0);props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);props.put(ProducerConfig.LINGER_MS_CONFIG, 1);props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return props;}@Beanpublic ProducerFactory<String, String> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigs());}public Map<String, Object> consumerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers);//props.put(ConsumerConfig.GROUP_ID_CONFIG, "0");props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "20000");props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);return props;}@Beanpublic ConsumerFactory<String, String> consumerFactory() {return new DefaultKafkaConsumerFactory<>(consumerConfigs());}@BeanConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory =new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());return factory;}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());return kafkaTemplate;}
}

消费者客户端：

    @KafkaListener(topics = {"topic-demo"},groupId = "test1",properties = {"auto-offset-reset:latest", "enable.auto.commit:true"})public void listen(ConsumerRecord<String, String> consumerRecord) {log.info("consumer Received: " + consumerRecord);}

生产者发送消息：

@RestController
@RequiredArgsConstructor
@RequestMapping("producer")
public class ProducerController {private final KafkaTemplate<String, String> kafkaTemplate;@PostMapping(path = "/sendCommonMsg")public String sendCommonMsg(String topic, String msg) {ListenableFuture<SendResult<String, String>> hello_kafka = this.kafkaTemplate.send(topic, "hello kafka");SendResult<String, String> sendResult = hello_kafka.completable().join();System.out.println(sendResult);return "send topic: " + topic + ", msg: " + msg;}
}

发送测试：
在这里插入图片描述

消费者可以接收到消息：

consumer Received: ConsumerRecord(topic = topic-demo, partition = 0, leaderEpoch = 0, offset = 2, CreateTime = 1721720091071, serialized key size = -1, serialized value size = 5, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)

9.IDEA客户端工具

可以使用kafkalytic工具本地开发环境可视化操作kafka服务器，如查看topic，创建topic

在这里插入图片描述

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

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

相关文章

迈向通用人工智能：AGI的到来与社会变革展望

迈向通用人工智能：AGI的到来与社会变革展望

正文&#xff1a; 随着科技的飞速发展&#xff0c;通用人工智能&#xff08;AGI&#xff09;的来临似乎已不再遥远。近期&#xff0c;多位行业领袖和专家纷纷预测&#xff0c;AGI的到来时间可能比我们想象的要早。在这篇博客中&#xff0c;我们将探讨AGI的发展趋势、潜在影响以…
阅读更多...
Mysql的主从复制（重要）和读写分离（理论重要实验不重要）

Mysql的主从复制（重要）和读写分离（理论重要实验不重要）

一、主从复制&#xff1a;架构一般是一主两从。 1.主从复制的模式&#xff1a; mysql默认模式为异步模式&#xff1a;主库在更新完事务之后会立即把结果返回给从服务器&#xff0c;并不关心从库是否接收到以及从库是否处理成功。缺点&#xff1a;网络问题没有同步、防火墙的等…
阅读更多...
JAVA零基础小白自学日志——第二十二天

JAVA零基础小白自学日志——第二十二天

文章目录 1.接口的方法[1].先来说说接口的默认方法[2].接口的静态方法 2.接口与抽象类的区别 今日提要&#xff1a;接口的静态方法和默认方法&#xff0c;接口与抽象类的区别 1.接口的方法 首先我们需要明确的是接口是一个抽象方法集&#xff0c;那就会有人问&#xff0c;为啥…
阅读更多...
vue3-video-play 导入 以及解决报错

vue3-video-play 导入 以及解决报错

npm install vue3-video-play --save # 或者 yarn add vue3-video-play import Vue3VideoPlay from vue3-video-play; import vue3-video-play/dist/style.css; app.use(Vue3VideoPlay) <template><div id"main-container-part"><div class"al…
阅读更多...
git配置name和email

git配置name和email

git配置name和email 1、下载好git之后&#xff0c;右击git bash&#xff0c;使用git config --global --list 查看配置信息&#xff0c;会出现以下错误 $ git config --global --list fatal: unable to read config file C:/Users/xxx/.gitconfig: No such file or directory…
阅读更多...
MySQL常见指令

MySQL常见指令

MySQL中的数据类型 大致分为五种&#xff1a;数值&#xff0c;日期和时间&#xff0c;字符串&#xff0c;json&#xff0c;空间类型 每种类型也包括也一些不同的子类型&#xff0c;根据需要来选择。 如数值类型包括整数类型和浮点数类型 整数类型根据占用的存储空间的不同 又…
阅读更多...
spice qxl-dod windows驱动笔记1

spice qxl-dod windows驱动笔记1

KMOD驱动是微软提供的一个Display Only驱动。 Windows驱动的入口函数是 DriverEntry ,所以显示Mini小端口驱动程序也不例外。 和其它Mini小端口驱动的入口函数实现一致&#xff0c;在其 DriverEntry 只做一件事&#xff0c;就是分配系统指定的一个结构体&#xff0c;然后调用框…
阅读更多...
Github遇到的问题解决方法总结（持续更新...）

Github遇到的问题解决方法总结（持续更新...）

1.github每次push都需要输入用户名和token的解决方法 push前&#xff0c;执行下面命令 &#xff1a; git config --global credential.helper store 之后再输入一次用户名和token之后&#xff0c;就不用再输入了。 2.git push时遇到“fatal: unable to access https://githu…
阅读更多...
Meta发布最强AI模型，扎克伯格公开信解释为何支持开源？

Meta发布最强AI模型，扎克伯格公开信解释为何支持开源？

凤凰网科技讯 北京时间7月24日&#xff0c;脸书母公司Meta周二发布了最新大语言模型Llama 3.1&#xff0c;这是该公司目前为止推出的最强大开源模型&#xff0c;号称能够比肩OpenAI等公司的私有大模型。与此同时&#xff0c;Meta CEO马克扎克伯格(Mark Zuckerberg)发表公开信&a…
阅读更多...
Spring Boot + Shiro 实现 Session 持久化实现思路及遗留问题

Spring Boot + Shiro 实现 Session 持久化实现思路及遗留问题

目录 引言 项目场景 应用技术 实现思路 问题暴露 解决方案 本人理解 引言 Session 为什么需要持久化？ Session 持久化的应用场景很多，诸如： 满足分布式：Session 作为有状态会话，体现在 Sessionid 与生成 Session 的服务器参数相关，在实现机理上不支持分布式部署…
阅读更多...
opencv grabCut前景后景分割去除背景

opencv grabCut前景后景分割去除背景

参考&#xff1a; https://zhuanlan.zhihu.com/p/523954762 https://docs.opencv.org/3.4/d8/d83/tutorial_py_grabcut.html 环境本次&#xff1a; python 3.10 提取前景&#xff1a; 1、需要先把前景物体框出来 需要坐标信息&#xff0c;可以用windows自带的画图简单提取像素…
阅读更多...
Concat() Function-SQL-字符串拼接函数

Concat() Function-SQL-字符串拼接函数

Concat() Function-SQL 在SQL中&#xff0c;CONCAT() 函数用于将两个或多个字符串连接在一起。 不同数据库管理系统可能有些许差异&#xff0c;但基本用法和语法通常是相似的。 语法 CONCAT(string1, string2, ...)string1, string2, …: 这些是需要连接的字符串参数。可以…
阅读更多...
089、Python 读取Excel文件及一些操作（使用openpyxl库）

089、Python 读取Excel文件及一些操作（使用openpyxl库）

对于低版本的Excel文件&#xff0c;我们可以使用xlwt/xlrd库&#xff0c;对于高版本的Excel文件(.xlsx)&#xff0c;xlwt/xlrd库从版本2.0.0开始不再支持&#xff0c;所以要读取.xlsx文件&#xff0c;我们需要单独使用openpyxl第三方库。 首先是安装&#xff1a; pip install…
阅读更多...
【时序约束】读懂用好Timing_report

【时序约束】读懂用好Timing_report

一、静态时序分析&#xff1a; 静态时序分析&#xff08;Static Timing Analysis&#xff09;简称 STA&#xff0c;采用穷尽的分析方法来提取出整个电路存在的所有时序路径&#xff0c;计算信号在这些路径上的传播延时&#xff0c;检查信号的建立和保持时间是否满足时序要求&a…
阅读更多...
Java并发编程实战读书笔记（二）

Java并发编程实战读书笔记（二）

对象的组合 在设计线程安全的类时&#xff0c;确保数据的一致性和防止数据竞争是至关重要的。这通常涉及三个基本要素&#xff1a;确定构成对象状态的所有变量&#xff0c;明确约束这些状态变量的不变性条件&#xff0c;以及建立管理对象状态并发访问的策略。 要确定构成对象…
阅读更多...
定时器+外部中断实现NEC红外线协议解码

定时器+外部中断实现NEC红外线协议解码

一、前言 1.1 功能介绍 随着科技的进步和人们生活水平的提高&#xff0c;红外遥控器已经成为了日常生活中不可或缺的电子设备之一&#xff0c;广泛应用于电视、空调、音响等多种家电产品中。 传统的红外遥控器通常只能实现预设的有限功能&#xff0c;无法满足用户对设备更加智…
阅读更多...
创建vue2/vue3项目

创建vue2/vue3项目

目录 创建一个Vue2项目创建一个Vue3项目 创建一个Vue2项目 ## 安装Vue-Cli &#xff1a; npm install -g vue/cli // Vue CLI 4.x 需要 Node.js v8.9 或更高版本 (推荐 v10 以上)vue --version // 检测版本是否正确## 创建一个项目&#xff1a; vue create hello-world // hel…
阅读更多...
TCP客户端connect断线重连

TCP客户端connect断线重连

文章目录 TCP客户端connect断线重连1、为什么要断线重连2、实现代码 TCP客户端connect断线重连 1、为什么要断线重连 客户端会面临服务器崩溃的情况&#xff0c;我们可以试着写一个客户端重连的代码&#xff0c;模拟并理解一些客户端行为&#xff0c;比如游戏客户端等. 考虑到…
阅读更多...
实战篇(十二)：如何使用 Processing 创建一个多功能的简易吃豆人游戏

实战篇(十二)：如何使用 Processing 创建一个多功能的简易吃豆人游戏

如何使用 Processing 创建一个多功能的简易吃豆人游戏 文章目录 如何使用 Processing 创建一个多功能的==简易==吃豆人游戏引言准备工作第一步：设置基本框架第二步：创建 Pacman 类第三步：创建 Obstacle 类第四步：添加分数系统第五步：运行游戏完整代码结论参考资料引言 吃…
阅读更多...
c++中的哈希查找（Hash Search）和B树查找（B-Tree Search）

c++中的哈希查找（Hash Search）和B树查找（B-Tree Search）

前言 hello大家好啊&#xff0c;我是文宇&#xff0c;不是文字&#xff0c;是文宇哦&#xff0c;这期也是关于查找算法的。 哈希查找&#xff08;Hash Search&#xff09; 哈希查找&#xff08;Hash Search&#xff09;是一种基于哈希表的查找算法&#xff0c;它可以在常数时…
阅读更多...
最新文章

Copyright @ 2022~2023