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

Spring Boot学习(三十三):集成kafka

前言

下面是zookeeper和kafka的官网下载地址,大家可以学习下载

zookeeper下载地址:http://zookeeper.apache.org/releases.html

kafka下载地址:http://kafka.apache.org/downloads.html

1、添加依赖

在 pom.xml 文件中添加kafka依赖,依赖如下

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

2、配置Kafka信息

在 application.properties(或 application.yml)文件中配置 Kafka 的相关信息,下面是一个简单的示例:

#kafka地址,多个地址使用,分隔
spring.kafka.bootstrap-servers=127.0.0.1:9092
#消费者组ID
spring.kafka.consumer.group-id=myGroup
#序列化和反序列化
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

3、发送消息

因为我们是springboot项目,已经集成了KafkaTemplate,我们可以直接使用KafkaTemplate来发送消息

下面,我编写一个发送消息的生产者

/*** 消息生产者*/
@Component
@Slf4j
public class KafkaProducer {@Autowiredprivate KafkaTemplate<String,String> kafkaTemplate;/*** 发送消息* @param topic 主题* @param msg   消息*/public void send(String topic,String msg){kafkaTemplate.send(topic,msg).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {@Overridepublic void onFailure(Throwable ex) {log.error("发送消息失败:{}", ex);}@Overridepublic void onSuccess(SendResult<String, String> result) {log.info("发送消息成功:{}");}});}/*** 发送消息* @param topic* @param msg*/public void send(String topic, Object msg) {send(topic, JSONObject.toJSONString(msg));}}

编写好生产者之后,我们就可以使用生产者发送消息,如下

	@Autowiredprivate KafkaProducer kafkaProducer;@GetMapping("send")public void sendMsg(){kafkaProducer.send("my-topic","hello world");}

如果想定制KafkaTemplate,那么可以在配置类进行配置,如下所示

@Configuration
public class KafakaConfig {@Value("${spring.kafka.bootstrap-servers}")private String bootstrapServers;/*** 配置属性* @return*/@Beanpublic Map<String, Object> producerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");return props;}@Beanpublic ProducerFactory<String, String> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigs());}/*** 定制KafkaTemplate* @return*/@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());kafkaTemplate.setDefaultTopic("myGroup");return kafkaTemplate;}}

4、消费消息

使用 @KafkaListener 注解创建 Kafka 消费者,并监听指定的主题。接收到消息后,可以通过方法参数来接收消息:

@Slf4j
@Component
public class KafkaConsumer {/*** 消费my-topic主题的消息* @param message*/@KafkaListener(topics = "my-topic",groupId = "myGroup1")public void  receiveMessage(String message){log.info("消费消息:"+message);}
}

同一消费者组,只会有一个消费者进行消费,如果想配置多个消费者同时处理,可以使用 @KafkaListener 注解来配置多个消费者。每个消费者需要配置不同的 group-id,监听主题一致,如下所示,就会有两个消费者同时消费

@Slf4j
@Component
public class KafkaConsumer {@KafkaListener(topics = "my-topic",groupId = "myGroup1")public void  receiveMessage(String message){log.info("消费消息:"+message);}@KafkaListener(topics = "my-topic",groupId = "myGroup2")public void  receiveMessage2(String message){log.info("消费消息:"+message);}}

相关文章:

Spring Boot学习(三十三):集成kafka

前言 下面是zookeeper和kafka的官网下载地址&#xff0c;大家可以学习下载 zookeeper下载地址&#xff1a;http://zookeeper.apache.org/releases.html kafka下载地址&#xff1a;http://kafka.apache.org/downloads.html 1、添加依赖 在 pom.xml 文件中添加kafka依赖&am…...

MOSFET

MOSFET 电子元器件百科 文章目录 MOSFET前言一、MOSFET是什么二、MOSFET类别三、MOSFET应用实例四、MOSFET作用原理总结前言 MOSFET是一种常见的半导体器件,通过栅极电场控制通道区的导通特性,以控制电流流动。它在现代电子电路中发挥着重要的作用,并广泛应用于各种应用领域…...

DriveWorks——参数化设计非标定制利器

DriveWorks基本介绍 DriveWorks是一套被 SOLIDWORKS 认可为金牌合作伙伴产品的设计自动化软件。DriveWorks 可自动创建特定于订单的销售文档和 SOLIDWORKS 制造数据。减少重复性任务&#xff0c;消除错误&#xff0c;增加销售额&#xff0c;并在创纪录的时间内交付定制产品。 为…...

DevEco Studio集成ArkUI-X

语雀知识库地址&#xff1a;语雀HarmonyOS知识库 飞书知识库地址&#xff1a;飞书HarmonyOS知识库 在上篇文章(HarmonyOS应用开发工具DevEco Studio安装与使用)中我说到官方推出了4.0 Beta版本的IDE&#xff0c;这篇文章就来介绍这个版本的安装与使用 该版本集成了HarmonyOS多…...

网络视频服务器的作用是什么?

随着互联网的快速发展和网络带宽的提升&#xff0c;网络视频已经成为人们日常生活中不可或缺的一部分。网络视频服务器作为支持和传输网络视频的关键基础设施&#xff0c;发挥着重要的作用。本文将以网络视频服务器的作用为方向&#xff0c;探讨其在现代社会中的重要性。 首先…...

解决vue3使用iconpark控制台预警提示问题

前言 最近在项目中使用 iconpark-icon 来管理图标&#xff0c;一切都很顺利&#xff0c;引入链接后&#xff0c;图标正常显示&#xff0c;没有报错。但是控制台却发出了预警信息。 [Vue warn]: Failed to resolve component: iconpark-icon If this is a native custom eleme…...

VMware 虚拟机 NAT 模式网络配置

配置的核心点在于 网关要一致&#xff0c;才能访问外网 比如下面的网关都是&#xff1a;192.168.145.2 问题总结&#xff1a; 当时重启电脑后如果连不上外网了&#xff0c;检查下 windows 服务中 NAT服务是否已经启动...

5-redis高级-哨兵

1 哨兵 1.1 python 操作哨兵 1 哨兵 # 主从---》一主多从-主库用来写-从库用来读-主库挂了--》整个系统就不能写数据了#主从复制存在的问题&#xff1a;1 主从复制&#xff0c;主节点发生故障&#xff0c;需要做故障转移&#xff0c;可以手动转移&#xff1a;让其中一个slave变…...

鸿蒙HarmonyOS4.0开发应用学习笔记

黑马程序员鸿蒙4.0视频学习笔记&#xff0c;供自己回顾使用。1.安装开发工具DevEco Studio 鸿蒙harmony开发文档指南 DevEco Studio下载地址 选择或者安装环境 选择和下载SDK 安装总览 编辑器界面 2.TypeScript语法 2.1变量声明 //string 、number、boolean、any、u…...

联通宽带+老毛子Padavan固件 开启IP v6

联通宽带开启IP v6 参考&#xff1a; 联通宽带开启 IPV6 的方法_联通ipv6怎么开通-CSDN博客 个人宽带如何开启IPv6网络访问 - 知乎 (zhihu.com) 首先&#xff0c;你要确定当前你所在的地区运营商已经开通了IPV6&#xff0c;可以使用手机流量 IP查询(ipw.cn) | IPv6测试 | IPv…...

唯创知音WT2003Hx系列单片机语音芯片:家庭理疗产品的智能声音伴侣

随着科技的不断创新&#xff0c;家庭理疗产品正迎来一场智能化的变革。唯创知音的WT2003Hx系列单片机语音芯片以其强大的功能和高品质音频播放能力&#xff0c;为家庭理疗产品带来了更为智能、沉浸式的用户体验。 1. MP3高品质音频播放 WT2003Hx系列语音芯片支持高品质的MP3音…...

2023_Spark_实验二十七:Linux中Crontab(定时任务)命令详解及使用教程

Crontab介绍&#xff1a; Linux crontab是用来crontab命令常见于Unix和类Unix的操作系统之中&#xff0c;用于设置周期性被执行的指令。该命令从标准输入设备读取指令&#xff0c;并将其存放于“crontab”文件中&#xff0c;以供之后读取和执行。该词来源于希腊语 chronos(χρ…...

Java动态代理实现与原理详细分析

Java动态代理实现与原理详细分析 关于Java中的动态代理&#xff0c;我们首先需要了解的是一种常用的设计模式–代理模式&#xff0c;而对于代理&#xff0c;根据创建代理类的 时间点&#xff0c;又可以分为静态代理和动态代理。 1、代理模式 代理模式是常用的java设计模式&…...

[实践总结] 使用Apache HttpClient 4.x进行进行一次Http请求

使用Apache HttpClient 4.x进行进行一次Http请求 依赖 <!-- https://mvnrepository.com/artifact/org.apache.httpcomponents/httpclient --> <dependency><groupId>org.apache.httpcomponents</groupId><artifactId>httpclient</artifactI…...

易宝OA 两处任意文件上传漏洞复现

0x01 产品简介 易宝OA系统是一种专门为企业和机构的日常办公工作提供服务的综合性软件平台,具有信息管理、 流程管理 、知识管理(档案和业务管理)、协同办公等多种功能。 0x02 漏洞概述 易宝OA系统UploadFile、BasicService.asmx等接口处存在文件上传漏洞,未授权的攻击者可…...

echart饼图高亮颜色设置,数据为0时候,labelLine不显示

鼠标移上去高亮&#xff0c;颜色变浅&#xff0c;希望不改变颜色 在series.data中为各项设置itemStyle&#xff0c;官方设置不生效&#xff0c;不知原因&#xff0c;可能版本问题 itemStyle: {normal: { color: #DFEAFF, },emphasis: { color: #DFEAFF }},数据为0时候显示饼图…...

Kafka 的消息格式:了解消息结构与序列化

Kafka 作为一款高性能的消息中间件系统&#xff0c;其消息格式对于消息的生产、传输和消费起着至关重要的作用。本篇博客将深入讨论 Kafka 的消息格式&#xff0c;包括消息的结构、序列化与反序列化&#xff0c;以及一些常用的消息格式选项。通过更丰富的示例代码和深入的解析&…...

装箱 Box 数据类型

装箱是最简单直接的一种智能指针&#xff0c;它的类型是Box<T>。装箱使我们可以把数据存储到堆上&#xff0c;并在栈上保留一个指向堆数据的指针。装箱操作常常被用于下面的场景&#xff1a; 当你拥有一个无法在编译时确定大小的类型&#xff0c;但又想使用这个类型的值…...

多传感器融合SLAM在自动驾驶方向的初步探索的记录

1. VIO的不可观问题 现有的VIO都是解决的六自由度的问题, 但是对于行驶在路面上的车来说, 通常情况下不会有roll与z方向的自由度, 而且车体模型限制了不可能有纯yaw的变换. 同时由于IMU在Z轴上与roll, pitch上激励不足, 会导致IMU在初始化过程中尺度不准以及重力方向估计错误,…...

ffmpeg与opencv-python处理视频

安装 opencv pip install opencv-pythonFFmpeg 1.下载 FFmpeg 访问FFmpeg官方网站。选择 “Windows builds from gyan.dev” 链接&#xff0c;这会带您到一个包含最新版本 FFmpeg Windows 构建的页面。选择一个适合您系统的版本&#xff08;例如&#xff0c;32位或64位&…...

浏览器访问 AWS ECS 上部署的 Docker 容器(监听 80 端口)

✅ 一、ECS 服务配置 Dockerfile 确保监听 80 端口 EXPOSE 80 CMD ["nginx", "-g", "daemon off;"]或 EXPOSE 80 CMD ["python3", "-m", "http.server", "80"]任务定义&#xff08;Task Definition&…...

【网络】每天掌握一个Linux命令 - iftop

在Linux系统中&#xff0c;iftop是网络管理的得力助手&#xff0c;能实时监控网络流量、连接情况等&#xff0c;帮助排查网络异常。接下来从多方面详细介绍它。 目录 【网络】每天掌握一个Linux命令 - iftop工具概述安装方式核心功能基础用法进阶操作实战案例面试题场景生产场景…...

rknn优化教程(二)

文章目录 1. 前述2. 三方库的封装2.1 xrepo中的库2.2 xrepo之外的库2.2.1 opencv2.2.2 rknnrt2.2.3 spdlog 3. rknn_engine库 1. 前述 OK&#xff0c;开始写第二篇的内容了。这篇博客主要能写一下&#xff1a; 如何给一些三方库按照xmake方式进行封装&#xff0c;供调用如何按…...

MongoDB学习和应用(高效的非关系型数据库)

一丶 MongoDB简介 对于社交类软件的功能&#xff0c;我们需要对它的功能特点进行分析&#xff1a; 数据量会随着用户数增大而增大读多写少价值较低非好友看不到其动态信息地理位置的查询… 针对以上特点进行分析各大存储工具&#xff1a; mysql&#xff1a;关系型数据库&am…...

聊聊 Pulsar:Producer 源码解析

一、前言 Apache Pulsar 是一个企业级的开源分布式消息传递平台&#xff0c;以其高性能、可扩展性和存储计算分离架构在消息队列和流处理领域独树一帜。在 Pulsar 的核心架构中&#xff0c;Producer&#xff08;生产者&#xff09; 是连接客户端应用与消息队列的第一步。生产者…...

项目部署到Linux上时遇到的错误(Redis,MySQL,无法正确连接,地址占用问题)

Redis无法正确连接 在运行jar包时出现了这样的错误 查询得知问题核心在于Redis连接失败&#xff0c;具体原因是客户端发送了密码认证请求&#xff0c;但Redis服务器未设置密码 1.为Redis设置密码&#xff08;匹配客户端配置&#xff09; 步骤&#xff1a; 1&#xff09;.修…...

安卓基础(aar)

重新设置java21的环境&#xff0c;临时设置 $env:JAVA_HOME "D:\Android Studio\jbr" 查看当前环境变量 JAVA_HOME 的值 echo $env:JAVA_HOME 构建ARR文件 ./gradlew :private-lib:assembleRelease 目录是这样的&#xff1a; MyApp/ ├── app/ …...

如何更改默认 Crontab 编辑器 ?

在 Linux 领域中&#xff0c;crontab 是您可能经常遇到的一个术语。这个实用程序在类 unix 操作系统上可用&#xff0c;用于调度在预定义时间和间隔自动执行的任务。这对管理员和高级用户非常有益&#xff0c;允许他们自动执行各种系统任务。 编辑 Crontab 文件通常使用文本编…...

Go语言多线程问题

打印零与奇偶数&#xff08;leetcode 1116&#xff09; 方法1&#xff1a;使用互斥锁和条件变量 package mainimport ("fmt""sync" )type ZeroEvenOdd struct {n intzeroMutex sync.MutexevenMutex sync.MutexoddMutex sync.Mutexcurrent int…...

Python+ZeroMQ实战:智能车辆状态监控与模拟模式自动切换

目录 关键点 技术实现1 技术实现2 摘要&#xff1a; 本文将介绍如何利用Python和ZeroMQ消息队列构建一个智能车辆状态监控系统。系统能够根据时间策略自动切换驾驶模式&#xff08;自动驾驶、人工驾驶、远程驾驶、主动安全&#xff09;&#xff0c;并通过实时消息推送更新车…...