【Kafka】SpringBoot整合Kafka详细介绍及代码示例

Kafka介绍

Apache Kafka是一个分布式流处理平台。它最初由LinkedIn开发,后来成为Apache软件基金会的一部分,并在开源社区中得到了广泛应用。Kafka的核心概念包括Producer、Consumer、Broker、Topic、Partition和Offset。

  • Producer:生产者,负责将数据发送到Kafka集群。
  • Consumer:消费者,从Kafka集群中读取数据。
  • Broker:Kafka服务器实例,Kafka集群通常由多个Broker组成。
  • Topic:主题,数据按主题进行分类。
  • Partition:分区,每个主题可以有多个分区,用于实现并行处理和提高吞吐量。
  • Offset:偏移量,每个消息在其分区中的唯一标识。

使用场景

Kafka适用于以下场景:

  1. 日志收集:集中收集系统日志和应用日志,通过Kafka传输到大数据处理系统。
  2. 消息队列:作为高吞吐量、低延迟的消息队列系统。
  3. 数据流处理:实时处理数据流,用于实时分析、监控和处理。
  4. 事件源架构:将所有的变更事件存储在Kafka中,实现事件溯源和回放。
  5. 流数据管道:构建数据管道,连接数据源和数据存储系统。

Spring Boot整合Kafka 

项目结构

springboot-kafka
│
├── src
│   ├── main
│   │   ├── java
│   │   │   └── com.example.kafka
│   │   │       ├── KafkaApplication.java
│   │   │       ├── config
│   │   │       │   └── KafkaConfig.java
│   │   │       ├── producer
│   │   │       │   └── KafkaProducer.java
│   │   │       ├── consumer
│   │   │       │   └── KafkaConsumer.java
│   │   │       └── controller
│   │   │           └── KafkaController.java
│   │   └── resources
│   │       ├── application.yml
│   │       └── logback-spring.xml (可选)
│   └── test
│       └── java
│           └── com.example.kafka
│               └── KafkaApplicationTests.java
└── pom.xml

1. 创建Spring Boot项目并添加依赖

pom.xml
<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-kafka</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency>
</dependencies>

2. 配置Kafka

application.yml
spring:kafka:bootstrap-servers: localhost:9092consumer:group-id: my-groupauto-offset-reset: earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializerproducer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer

3. 创建Kafka配置类

KafkaConfig.java
package com.example.kafka.config;import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class KafkaConfig {@Beanpublic NewTopic myTopic() {return new NewTopic("my-topic", 1, (short) 1);}
}

4. 创建Kafka生产者

KafkaProducer.java
package com.example.kafka.producer;import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;@Service
public class KafkaProducer {private final KafkaTemplate<String, String> kafkaTemplate;public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {this.kafkaTemplate = kafkaTemplate;}public void sendMessage(String topic, String message) {kafkaTemplate.send(topic, message);}
}

5. 创建Kafka消费者

KafkaConsumer.java
package com.example.kafka.consumer;import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;@Service
public class KafkaConsumer {@KafkaListener(topics = "my-topic", groupId = "my-group")public void listen(String message) {System.out.println("Received message: " + message);}
}

6. 创建控制器发送消息

KafkaController.java
package com.example.kafka.controller;import com.example.kafka.producer.KafkaProducer;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;@RestController
public class KafkaController {private final KafkaProducer kafkaProducer;public KafkaController(KafkaProducer kafkaProducer) {this.kafkaProducer = kafkaProducer;}@GetMapping("/send")public String sendMessage(@RequestParam String message) {kafkaProducer.sendMessage("my-topic", message);return "Message sent";}
}

7. 创建Spring Boot主类

KafkaApplication.java
package com.example.kafka;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class KafkaApplication {public static void main(String[] args) {SpringApplication.run(KafkaApplication.class, args);}
}

8. 测试应用

通过访问以下URL来发送消息:

http://localhost:8080/send?message=HelloKafka

9. 日志配置(可选)

为了更好地查看Kafka的日志,可以添加logback-spring.xml配置:

logback-spring.xml
<configuration><springProfile name="default"><appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"><encoder><pattern>%d{yyyy-MM-dd HH:mm:ss} - %msg%n</pattern></encoder></appender><logger name="org.apache.kafka" level="INFO"/><root level="INFO"><appender-ref ref="STDOUT"/></root></springProfile>
</configuration>

10. 测试类(可选)

KafkaApplicationTests.java
package com.example.kafka;import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;@SpringBootTest
class KafkaApplicationTests {@Testvoid contextLoads() {}
}

至此,你已经完成了Spring Boot整合Kafka的详细配置和代码示例。你可以根据实际需求进一步扩展和修改这个基础代码。

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

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

相关文章

车载android开发 carservice(一)

车载android开发 carservice是什么? 车载Android开发中的CarService是一个专门为汽车环境设计的系统服务。CarService通常是Android Automotive OS的一部分,提供一系列API和框架,允许开发人员构建与汽车相关的应用和服务。 以下是CarService的一些主要功能和作用: 车辆数…

宽睿数字平台兼容TDengine 等多种数据库,提供行情解决方案

小T导读&#xff1a;最近&#xff0c;涛思数据与宽睿金融宣布了一项重要合作。在此之前&#xff0c;宽睿金融对 TDengine 进行了性能测试&#xff0c;并根据测试报告的结果&#xff0c;决定将 TDengine 接入宽睿数字平台&#xff0c;以提升高密度行情处理效率。本文将详细介绍此…

百元开放式耳机性价比推荐,全网5款爆火产品实测!

耳机作为我们日常生活中不可或缺的一部分&#xff0c;它的性能、配置及形态也在不停的优化。目前开放式耳机是耳机中最为热门的一款产品&#xff0c;以其不入耳的的设计迅速占领耳机市场&#xff0c;最主要是因为自身佩戴的舒适度有大幅度提升。人们更加注重生活的质量&#xf…

Apollo9.0 PNC源码学习之Control模块(五)—— 基于LQR的横向控制

前面文章&#xff1a; Apollo9.0 PNC源码学习之Control模块&#xff08;一&#xff09; Apollo9.0 PNC源码学习之Control模块&#xff08;二&#xff09; Apollo9.0 PNC源码学习之Control模块&#xff08;三&#xff09; Apollo9.0 PNC源码学习之Control模块&#xff08;四&…

通信原理抽样定理和PAM调制解调硬件实验

一、实验目的 1. 加深理解抽样定理&#xff1b; 2. 加深理解脉冲幅度调制的原理。 二、实验内容 1. 观测PAM平顶抽样波形&#xff1b; 2. 观测PAM自然抽样波形及解码后波形。 三、实验器材 1. 双踪示波器&#xff1b; 2. 通信原理实验箱信号源模块、①号模块。 四、实…

Stable Diffusion学习指南【图生图篇】

一、图生图功能初识 1.1 传统意义上的喂参考图 我们都知道&#xff0c;模型在运算时是根据我们提供的提示内容来确定绘图方向&#xff0c;如果没有提示信息&#xff0c;模型只能根据此前的学习经验来自行发挥。在之前的文生图篇&#xff0c;我们介绍了如何通过提示词来控制图…

Centos系统yum安装mysql数据库

安装之前需要将系统自带的mariadb-libs软件包删除。 检查是否存在mariadb-libs包。 yum list installed|grep mariadb-libs 删除mariadb-libs包 yum -y remove mariadb-libs 声明&#xff1a; 系统&#xff1a;CentOS-7-x86_64-DVD-2009 安装为最小化安装&#xff0c;没…

在IDEA中连接数据库

1.点击侧边栏的Database 2.选中你要连接的数据库类型 3.输入用户名、密码&#xff0c;测试连接&#xff0c;没有驱动的话先下载下驱动 4.发现没有数据库&#xff0c;点击1 of 14&#xff0c;选择All schemas&#xff0c;即可显示全部数据库 5.执行sql 如果对您有帮助的话记…

【ASE】笔记总结

这里写自定义目录标题 ASE 介绍ASE 的安装文件格式的转化命令行操作图形界面操作Python代码操作[2]支持的文件转化格式 ASE&#xff08;三&#xff09;&#xff1a;图形界面查看/建立/修改结构图形界面召唤查看结构 python代码查看/建立结构读写、查看结构建立结构分子 周期性体…

深入浅出MyBatis:全面解析与实战指南

MyBatis 是一个优秀的持久层框架&#xff0c;它简化了 Java 应用与关系数据库之间的映射。对于大多数 Java 开发者而言&#xff0c;掌握 MyBatis 是必不可少的一部分。本文将详细介绍 MyBatis 的各个方面&#xff0c;包括其基本原理、配置、操作、动态 SQL、插件机制和高级应用…

新创作活动,The Sandbox 宣布第二届 “创作者挑战”!

是时候展示你的实力了&#xff01; 继首届 “创作者挑战” 成功举办之后&#xff0c;The Sandbox 很高兴推出第二届 "创作者挑战"&#xff0c;邀请创作者再次释放他们的创造力和创业精神。本次活动从 2024 年 6 月 24 日开始至 8 月 26 日结束&#xff0c;为期 10 …

Ant-design-vue开源项目介绍、应用场景、组件有哪些

文章目录 一、Ant-design-vue项目介绍二、Ant-design-vue项目特点三、Ant-design-vue应用场景四、Ant-design-vue有哪些组件五、Ant-design-vue案例代码1. 后台管理系统登录页面的例子2. Table组件使用案例 开源项目地址 一、Ant-design-vue项目介绍 Ant-design-vue 是一个基于…

第104天: 权限提升-Linux 系统环境变量定时任务权限配置不当MDUT 自动化

目录 案例一&#xff1a;Linux-环境变量文件配合 SUID-本地 案例二&#xff1a;Linux-定时任务打包配合 SUID-本地 案例三&#xff1a;Linux-定时任务文件权限配置不当-WEB&本地 案例四&#xff1a;Linux-第三方软件 MYSQL 数据库提权-WEB&本地 隧道出网 手工提权…

Science Online《科学周刊》文献在家如何查找下载

今天有位同学自己学校没有Science《科学》数据库&#xff0c;求助几篇该数据库文献&#xff0c;小编就把自己在家下载文献的方法分享给了这位同学&#xff0c;同时也分享给其他有需要文献下载但找不到资源的朋友。本文以下载Science《科学》文献为例来演示一下在家获取文献的过…

STM32开发遇到问题解决2024.6.14

1、电路板打样出问题 电路板样品回来&#xff0c;做了接线&#xff0c;连接上后&#xff0c;可以下载程序&#xff0c;上电测试&#xff0c;有画面&#xff0c;但是上电不到一分钟&#xff0c;电池电量突增&#xff0c;串口屏开始闪烁&#xff0c;3.3V指示灯闪烁&#xff0c;赶…

leetcode刷题记录40-886. 可能的二分法

问题描述 给定一组 n 人&#xff08;编号为 1, 2, ..., n&#xff09;&#xff0c; 我们想把每个人分进任意大小的两组。每个人都可能不喜欢其他人&#xff0c;那么他们不应该属于同一组。 给定整数 n 和数组 dislikes &#xff0c;其中 dislikes[i] [ai, bi] &#xff0c;表示…

32T存储删除视频的恢复方法

由于存储技术的发展和普及目前很多行业都开始使用小型存储&#xff0c;NAS可以通过网络进行数据上传和读取&#xff0c;使用极为方便。但是由于NAS设备容量较大且碎片较多&#xff0c;所以此类设备删除或者格式后恢复难度是比较大的&#xff0c;下边我们来分享下32T存储的恢复方…

opencv安装笔记 各种平台

目录 python安装opencv-python c 麒麟arm系统安装和用法 python安装opencv-python pypi上搜索 Search results PyPI 现在安装是一个版本&#xff0c;大于3.6都可以安装 c 麒麟arm系统安装和用法 参考&#xff1a; ffmpeg rknn麒麟系统 安装 opencv_ffmpeg4 解码示例-CSDN…

【Gitlab】访问默认PostgreSQL数据库

本地访问PostgreSQL gitlab有可以直接访问内部PostgreSQL的命令 sudo gitlab-rails dbconsole # 或者 sudo gitlab-psql -d gitlabhq_production效果截图 常用SQL # 查看用户状态 select id,name,email,state,last_sign_in_at,updated_at,last_credential_check_at,last_act…

SqlSugar使用DbFirst对象根据数据库表结构创建实体类-C#

本文所述开发环境&#xff1a;.C#、NET8、Visual Studio2022 1. 在项目中安装SqlSugar 在Visual Studio2022中新建一个 C# 的控制台应用程序&#xff0c;框架选择 .Net8。新建后如下图所示&#xff1a; 然后打开NuGet程序包管理器 搜索 SqlSugarCore 并安装 安装后在解决方案…