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

【实战】Spring Cloud Stream 3.1+整合Kafka

文章目录

    • 前言
    • 新版版本优势
    • 实战演示
      • 增加maven依赖
      • 增加applicaiton.yaml配置
      • 新增Kafka通道消费者
      • 新增发送消息的接口
    • 实战测试
      • postman发送一个正常的消息
      • postman发送异常消息

前言

之前我们已经整合过Spring Cloud Stream 3.0版本与Kafka、RabbitMQ中间件,简直不要太好,直接让我们不用再关心底层MQ如何集与消息收发。但是从Spring Cloud 2020版本开始,Spring Cloud Stream的版本升级至3.1.0以上版本,自此版本开始@StreamListener上面就增加@Deprecated注解,不赞成使用,有可能接下来的版本会删除掉。传说是有利于使用Project Reactor提供的事件流抽象(如Flux和Mono),命令函数在每个单独的事件上触发,而reactive函数只触发一次。故今天我们分享一期Spring Cloud Stream 3.1+整合Kafka,各位看官敬请鉴赏。

在这里插入图片描述

新版版本优势

新版提倡用函数式进行发送和消费信息

定义返回类型为Supplier, Function or Consumer的bean提供消息发送和消费的bean 看看绑定名称命名规则
input - + -in- +
output - + -out- +

在配置文件中指定spring.cloud.function.definition/spring.cloud.stream.function.definition的名称后会把这个bean绑定到对应的消费者和提供者上。

比如 inputChannel bean绑定了inputChannel-in-0通道,outputChannel bean绑定了outputChannel-out-0通道:

spring:kafka:bootstrap-servers: 192.168.112.10:9092,192.168.112.130:9092,192.168.112.129:9092cloud:stream:kafka:binder:brokers: ${spring.kafka.bootstrap-servers}binders:kafkahub:type: kafkaenvironment:spring:cloud:stream:kafka: ${spring.cloud.stream.kafka.binder}default-binder: kafkahub function:definition: inputChannel,outputChannelbindings:inputChannel-in-0:binder: kafkahubdestination: test-kafka-topicgroup: test-kafka-groupcontent-type: text/plainoutputChannel-out-0:binder: kafkahubdestination: test-kafka-topiccontent-type: text/plainproducer:partition-count: 3 #分区数目

此时消息生产者为:

@Resource
private StreamBridge streamBridge;@GetMapping("/send")
public Boolean sendMessageToKafka(String msg){boolean send = streamBridge.send("outputChannel-out-0", MessageBuilder.withPayload("kafka测试:"+msg).build());return send;
}

此时消息消费者为:

@Configuration
public class KafkaChannel {@Resourceprivate StreamBridge streamBridge;/*** inputChannel 消费者* @author senfel* @date 2024/6/18 15:26* @return java.util.function.Consumer<java.lang.String>*/@Beanpublic Consumer<Message<String>> inputChannel(){return message -> {System.out.println("接收到消息Payloa:" + message.getPayload());System.out.println("接收到消息Header:" + message.getHeaders());};}

}

实战演示

我们简单进行一下演示即可,kafka环境可以看我之前的博文搭建。
主要演示功能:

正常情况下生产者发送消息到kafka,消费者监听消息并消费成功
异常情况下消费者消费失败,立即将异常消息投递到另一个topic上,兜底topic消费者消费

本次全部采用自动ack模式,如果需要手动ack参照之前的博文配置即可,注意在消费者端加上手动ack逻辑。

增加maven依赖

 <parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.12.RELEASE</version><relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>cce-demo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>seata-demo-order</name>
<description>Demo project for Spring Boot</description>
<properties><java.version>8</java.version><spring-cloud.version>Hoxton.SR12</spring-cloud.version>
</properties>
<dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-kafka</artifactId><version>3.2.4</version></dependency>
</dependencies>
<dependencyManagement><dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-dependencies</artifactId><version>${spring-cloud.version}</version><type>pom</type><scope>import</scope></dependency></dependencies>
</dependencyManagement>

增加applicaiton.yaml配置

spring:#kafkakafka:bootstrap-servers: 192.168.112.10:9092,192.168.112.130:9092,192.168.112.129:9092cloud:stream:kafka:  # kafka配置binder:brokers: ${spring.kafka.bootstrap-servers}auto-add-partitions: true #自动分区auto-create-topics: true #自动创建主题replication-factor: 3 #副本min-partition-count: 3 #最小分区bindings:outputChannel-out-0:producer:# 无限制重发不产生消息丢失retries: Integer.MAX_VALUE#acks =0:producer不等待broker的ack,broker一接收到还没有写入磁盘就已经返回,可靠性最低#acks =1:producer等待broker的ack,partition的leader刷盘成功后返回ack,如果在follower同步成功之前leader故障,那么将会丢失数据,可靠性中#acks = all 、 -1:producer等待broker的ack,partition的leader和follower全部落盘成功后才返回ack,可靠性高,但延迟时间长#可以设置的值为:all, -1, 0, 1acks: allmin:insync:replicas: 3 #感知副本数inputChannel-in-0:consumer:concurrency: 1 #消费者数量max-concurrency: 5 #最大消费者数量recovery-interval: 3000  #3s 重连auto-rebalance-enabled: true  #主题分区消费者组成员自动平衡auto-commit-offset: false   #手动提交偏移量enable-dlq: true  # 开启 dlq队列dlq-name: test-kafka-topic.dlqdeserializationExceptionHandler: sendToDlq #异常加入死信binders: # 与外部mq组件绑定kafkahub:type: kafkaenvironment:spring:cloud:stream:kafka: ${spring.cloud.stream.kafka.binder}default-binder: kafkahub #默认绑定function: # 定义channel名字,每个channel又可以作为生产者(in)与消费者(out)definition: inputChannel;outputChannel;dlqChannelbindings: # 通道绑定inputChannel-in-0:binder: kafkahubdestination: test-kafka-topicgroup: test-kafka-groupcontent-type: text/plainconsumer:maxAttempts: 1 # 尝试消费该消息的最大次数(消息消费失败后,发布者会重新投递)。默认3backOffInitialInterval: 1000 # 重试消费消息的初始化间隔时间。默认1s,即第一次重试消费会在1s后进行backOffMultiplier: 2 # 相邻两次重试之间的间隔时间的倍数。默认2backOffMaxInterval: 10000 # 下一次尝试重试的最大时间间隔,默认为10000ms,即10soutputChannel-out-0:binder: kafkahubdestination: test-kafka-topiccontent-type: text/plainproducer:partition-count: 3 #分区数目dlqChannel-in-0:binder: kafkahubdestination: test-kafka-topic.dlqgroup: test-kafka-groupcontent-type: text/plainconsumer:maxAttempts: 1 # 尝试消费该消息的最大次数(消息消费失败后,发布者会重新投递)。默认3backOffInitialInterval: 1000 # 重试消费消息的初始化间隔时间。默认1s,即第一次重试消费会在1s后进行backOffMultiplier: 2 # 相邻两次重试之间的间隔时间的倍数。默认2backOffMaxInterval: 10000 # 下一次尝试重试的最大时间间隔,默认为10000ms,即10sdlqChannel-out-0:binder: kafkahubdestination: test-kafka-topic.dlqcontent-type: text/plainproducer:partition-count: 3 #分区数目

新增Kafka通道消费者

import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import javax.annotation.Resource;
import java.util.function.Consumer;/*** KafkaCustomer* @author senfel* @version 1.0* @date 2024/6/18 15:22*/
@Configuration
public class KafkaChannel {@Resourceprivate StreamBridge streamBridge;/*** inputChannel 消费者* @author senfel* @date 2024/6/18 15:26* @return java.util.function.Consumer<java.lang.String>*/@Beanpublic Consumer<Message<String>> inputChannel(){return message -> {System.out.println("接收到消息:" + message.getPayload());System.out.println("接收到消息:" + message.getHeaders());if(message.getPayload().contains("9")){boolean send = streamBridge.send("dlqChannel-out-0", MessageBuilder.withPayload("kafka异常消息发送到dlq测试:"+message).build());System.err.println("向dlqChannel发送消息:"+send);}};}/*** dlqChannel 死信消费者* @author senfel* @date 2024/6/18 15:26* @return java.util.function.Consumer<java.lang.String>*/@Beanpublic Consumer<Message<String>> dlqChannel(){return message -> {System.out.println("死信dlqChannel接收到消息:" + message.getPayload());System.out.println("死信dlqChannel接收到消息:" + message.getHeaders());};}
}

新增发送消息的接口

@Resource
private StreamBridge streamBridge;@GetMapping("/send")
public Boolean sendMessageToKafka(String msg){boolean send = streamBridge.send("outputChannel-out-0", MessageBuilder.withPayload("kafka测试:"+msg).build());return send;
}

实战测试

postman发送一个正常的消息

在这里插入图片描述

postman发送异常消息

相关文章:

【实战】Spring Cloud Stream 3.1+整合Kafka

文章目录 前言新版版本优势实战演示增加maven依赖增加applicaiton.yaml配置新增Kafka通道消费者新增发送消息的接口 实战测试postman发送一个正常的消息postman发送异常消息 前言 之前我们已经整合过Spring Cloud Stream 3.0版本与Kafka、RabbitMQ中间件&#xff0c;简直不要太…...

java之可变字符串之append方法

可变字符串如果要添加内容&#xff0c;需要用到append方法 语法格式如下 sbf.append(obj) 其中sbf是任意的可变字符串 obj是任意数据类型的对象 这个方法是将任意数据转换成字符串&#xff0c;然后添加到此序列中 public class Buffer {public static void main(String[]…...

[保姆级教程]uniapp自定义导航栏

文章目录 导文隐藏默认导航栏&#xff1a;全局隐藏当前页面隐藏 添加自定义导航栏视图&#xff1a;手写导航栏组件导航栏 导文 在 UniApp 中&#xff0c;自定义导航栏通常涉及到隐藏默认的导航栏&#xff0c;并在页面顶部添加自定义的视图组件来模拟导航栏的功能。 隐藏默认导航…...

项目训练营第二天

项目训练营第二天 用户登录逻辑 1、账户名不少于4位 2、密码不少于8位 3、数据库表中能够查询到账户、密码 4、密码查询时用同样加密脱敏处理手段处理后再和数据库中取出字段进行对比&#xff0c;如果账户名未查询到&#xff0c;直接返回null 5、后端设置相应的脱敏后用户的s…...

考研数学一有多难?130+背后的残酷真相

考研数学一很难 大家平时在网上上看到很多人说自己考了130&#xff0c;其实这些人只占参加考研数学人数的极少部分&#xff0c;有个数据可以展示出来考研数学到底有多难&#xff1a; 在几百万考研大军中&#xff0c;能考到120分以上的考生只有2%。绝大多数人的分数集中在30到…...

vue2脚手架笔记总结1

1.什么是组件 组件是实现局部代码和功能资源的集合 2.vue.config.js配置文件 使用vue inspect > output.js可以查看到Vue脚手架的默认配置,但是在这里面修改不会影响实际的配置&#xff0c;如果需要修改配置需要使用用vue.config.js文件&#xff0c;详情见:https://cli.vuej…...

校园巡礼:一周只上四天课,入学即发钱?深圳理工大学,开局即王炸

校园巡礼 | 一周只上四天课&#xff0c;入学即发钱&#xff1f;深圳理工大学&#xff0c;开局即王炸&#xff01; 会议之眼 快讯 目前各省的高考成绩现已陆续揭晓&#xff0c;广东省教育考试院发布了2024年高考录取最低分数线&#xff0c;物理类本科线为442分&#xff0c;历史…...

免交互 实验

免交互 交互&#xff1a;我们发出指令控制程序的运行&#xff0c;程序在接收到指令之后按照指令的效果做出对应的反应。 免交互&#xff1a;间接的&#xff0c;通过第三方的方式把指令传送给程序&#xff0c;不用直接下达指令。 Here document 免交互 这是命令行格式&#…...

Sublime Text 设置

备份 {"font_size": 10,"index_files": true,"font_face": "Courier New","vintage_start_in_command_mode": false,"ignored_packages": ["Vintage"],"word_wrap": "false" }关闭…...

spire.Pdf 将pdf转成image

一、nuget安装 <ItemGroup><PackageReference Include"Spire.PDF" Version"10.6.7" /></ItemGroup> 二、直接上代码 using Microsoft.AspNetCore.Mvc; using Microsoft.Extensions.Logging; using System; using System.IO;namespace …...

仓颉编程语言 -- 初识(一)

官网 文档 原生智能化 内嵌AgentDSL的编程框架&#xff0c;自然语言&编程语言有机融合&#xff1b;多Agent协同&#xff0c;简化符号表达&#xff0c;模式自由组合&#xff0c;支持各类智能应用开发。 天生全场景 轻量化可缩放运行时&#xff0c;模块化分层设计&#xf…...

前端JS必用工具【js-tool-big-box】学习,数值型数组的正向排序和倒向排序

这一小节&#xff0c;我们说一下前端 js-tool-big-box 这个工具库&#xff0c;添加的数值型数组的正向排序和倒向排序。 以前呢&#xff0c;我们的数组需要排序的时候&#xff0c;都是在项目的utils目录里&#xff0c;写一段公共方法&#xff0c;弄个冒泡排序啦&#xff0c;弄…...

python web框架哪家强?Flask、Django、FastAPI对比

前言 当你掌握了python的基础知识&#xff0c;并且会用和HTML和CSS编写简单的静态网页。现在你只需再掌握一个python web框架的知识&#xff0c;就可以开始编写一个动态的网站了。目前市面比较流程的python web框架有三个flask、Django、FastAPI。接下来我们对比一下。他们三个…...

Mybatis plus:IService接口

一、介绍 在MybatisPlus框架中&#xff0c;IService接口扮演着重要的角色。作为一个通用的服务接口&#xff0c;IService定义了一系列方法&#xff0c;包括查询、插入、更新、删除等。这些方法的定义使得在服务层进行数据库操作变得更为便捷和高效。 IService 接口是一个泛型接…...

时序分析基本概念介绍——min pulse width 最小脉冲宽度

文章目录 前言一、什么是 min pulse width&#xff1f;二、为什么检查 min pulse width&#xff1f;三、如何设置 min pulse width约束&#xff1f;1. 在sdc里面定义2. library里面定义 四、如何检查 min pulse width&#xff1f;五、如何修复 min pulse width&#xff1f;总结…...

PHP原生代码生成pdf---解决中文乱码问题

github地址 尝试了使用composer下载FPDF或者FPDI&#xff0c;但是无法解决中文乱码问题。只有使用这个github上的中文包才可以&#xff0c;那俩没必要下。 直接上代码(这里并没有使用任何框架) require(./fpdf/chinese.php);//生成pdf$pdf new PDF_Chinese();$pdf->AddPage…...

智慧车库管理系统

摘 要 随着城市化进程的不断加快&#xff0c;私家车数量的快速增长给城市交通带来了巨大的挑战&#xff0c;停车问题成为城市交通管理中的一大难题。车辆停车时&#xff0c;在停车场寻找停车位耗时过久&#xff0c;不仅仅浪费用户的时间&#xff0c;还可能引起交通拥堵。城市停…...

每日新闻掌握【2024年6月26日 星期三】

2024年6月26日 星期三 农历五月廿一 大公司/大事件 OpenAI将终止对中国提供API服务 从6月24日晚间开始&#xff0c;已有多名用户收到了来自OpenAI的邮件。该邮件表示&#xff0c;“我们的数据显示您的组织来自OpenAI目前不支持的地区的API流量。”邮件进一步表示&#xff0c;…...

InVEST实践及在生态系统服务供需、固碳、城市热岛、论文写作等实际项目中应用

白老师&#xff08;研究员&#xff09;&#xff1a;长期从事生态系统结构-格局-过程-功能-服务的变化与响应关系等研究工作,重点围绕生物多样性、生态系统服务与价值等&#xff0c;构建生物地球化学模型和评价指标体系&#xff0c;为城市、区域和自然保护区的可持续发展和生态环…...

慧科新闻搜索研究数据库的使用指南及个人获取途径

《慧科新闻搜索研究数据库》WiseSearch由慧科讯业有限公司出品。WiseSearch是具有新闻搜索/浏览、对比分析等功能的一站式新闻搜索平台&#xff1b;内容包括1200种报刊和8000 网站的新闻资讯&#xff0c;平面媒体涵盖全国综合大报、党委机关报、都市报、行业报刊媒体&#xff0…...

PPT|230页| 制造集团企业供应链端到端的数字化解决方案:从需求到结算的全链路业务闭环构建

制造业采购供应链管理是企业运营的核心环节&#xff0c;供应链协同管理在供应链上下游企业之间建立紧密的合作关系&#xff0c;通过信息共享、资源整合、业务协同等方式&#xff0c;实现供应链的全面管理和优化&#xff0c;提高供应链的效率和透明度&#xff0c;降低供应链的成…...

使用分级同态加密防御梯度泄漏

抽象 联邦学习 &#xff08;FL&#xff09; 支持跨分布式客户端进行协作模型训练&#xff0c;而无需共享原始数据&#xff0c;这使其成为在互联和自动驾驶汽车 &#xff08;CAV&#xff09; 等领域保护隐私的机器学习的一种很有前途的方法。然而&#xff0c;最近的研究表明&…...

React Native在HarmonyOS 5.0阅读类应用开发中的实践

一、技术选型背景 随着HarmonyOS 5.0对Web兼容层的增强&#xff0c;React Native作为跨平台框架可通过重新编译ArkTS组件实现85%以上的代码复用率。阅读类应用具有UI复杂度低、数据流清晰的特点。 二、核心实现方案 1. 环境配置 &#xff08;1&#xff09;使用React Native…...

Vue2 第一节_Vue2上手_插值表达式{{}}_访问数据和修改数据_Vue开发者工具

文章目录 1.Vue2上手-如何创建一个Vue实例,进行初始化渲染2. 插值表达式{{}}3. 访问数据和修改数据4. vue响应式5. Vue开发者工具--方便调试 1.Vue2上手-如何创建一个Vue实例,进行初始化渲染 准备容器引包创建Vue实例 new Vue()指定配置项 ->渲染数据 准备一个容器,例如: …...

2025盘古石杯决赛【手机取证】

前言 第三届盘古石杯国际电子数据取证大赛决赛 最后一题没有解出来&#xff0c;实在找不到&#xff0c;希望有大佬教一下我。 还有就会议时间&#xff0c;我感觉不是图片时间&#xff0c;因为在电脑看到是其他时间用老会议系统开的会。 手机取证 1、分析鸿蒙手机检材&#x…...

【Java_EE】Spring MVC

目录 Spring Web MVC ​编辑注解 RestController RequestMapping RequestParam RequestParam RequestBody PathVariable RequestPart 参数传递 注意事项 ​编辑参数重命名 RequestParam ​编辑​编辑传递集合 RequestParam 传递JSON数据 ​编辑RequestBody ​…...

数据库分批入库

今天在工作中&#xff0c;遇到一个问题&#xff0c;就是分批查询的时候&#xff0c;由于批次过大导致出现了一些问题&#xff0c;一下是问题描述和解决方案&#xff1a; 示例&#xff1a; // 假设已有数据列表 dataList 和 PreparedStatement pstmt int batchSize 1000; // …...

ip子接口配置及删除

配置永久生效的子接口&#xff0c;2个IP 都可以登录你这一台服务器。重启不失效。 永久的 [应用] vi /etc/sysconfig/network-scripts/ifcfg-eth0修改文件内内容 TYPE"Ethernet" BOOTPROTO"none" NAME"eth0" DEVICE"eth0" ONBOOT&q…...

【Redis】笔记|第8节|大厂高并发缓存架构实战与优化

缓存架构 代码结构 代码详情 功能点&#xff1a; 多级缓存&#xff0c;先查本地缓存&#xff0c;再查Redis&#xff0c;最后才查数据库热点数据重建逻辑使用分布式锁&#xff0c;二次查询更新缓存采用读写锁提升性能采用Redis的发布订阅机制通知所有实例更新本地缓存适用读多…...

day36-多路IO复用

一、基本概念 &#xff08;服务器多客户端模型&#xff09; 定义&#xff1a;单线程或单进程同时监测若干个文件描述符是否可以执行IO操作的能力 作用&#xff1a;应用程序通常需要处理来自多条事件流中的事件&#xff0c;比如我现在用的电脑&#xff0c;需要同时处理键盘鼠标…...