kafka+spring cloud stream 发送接收消息
方案 1:使用旧版 @StreamListener(适用于 Spring Cloud Stream <= 2.x)
1. 添加依赖(pom.xml)
<!-- Spring Cloud Stream + Kafka Binder -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
2. 定义消息通道接口
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
public interface MyChannels {
String INPUT = "myInput";
@Input(INPUT)
SubscribableChannel input();
}
3. 使用 @StreamListener 监听
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;
@Service
@EnableBinding(MyChannels.class) // 绑定消息通道
public class KafkaStreamListener {
@StreamListener(MyChannels.INPUT)
public void handleMessage(String message) {
System.out.println("Received via @StreamListener: " + message);
}
}
4. 配置 application.properties
# Kafka 服务器地址
spring.kafka.bootstrap-servers=localhost:9092
# 绑定输入通道到 Kafka Topic
spring.cloud.stream.bindings.myInput.destination=my-topic
spring.cloud.stream.bindings.myInput.group=my-group
spring.cloud.stream.bindings.myInput.content-type=text/plain
方案 2:新版函数式编程模型(推荐,Spring Cloud Stream >= 3.x)
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import java.util.function.Consumer;
@Component
public class KafkaStreamListener {
@Bean
public Consumer<String> myInput() {
return message -> {
System.out.println("Received via Function: " + message);
};
}
}
# Kafka 服务器地址
spring.kafka.bootstrap-servers=localhost:9092
# 绑定函数到 Kafka Topic
spring.cloud.stream.bindings.myInput-in-0.destination=my-topic
spring.cloud.stream.bindings.myInput-in-0.group=my-group
spring.cloud.stream.bindings.myInput-in-0.content-type=text/plain
生产者代码示例(发送消息)
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Service;
@Service
public class KafkaProducer {
@Autowired
private StreamBridge streamBridge;
public void sendMessage(String topic, String message) {
streamBridge.send(topic, message);
}
}
测试步骤
-
启动 Kafka:确保 Kafka 和 Zookeeper 服务运行。
-
创建 Topic:
kafka-topics --create --topic my-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
-
发送消息:
kafkaProducer.sendMessage("my-topic", "Hello Kafka Stream!");
-
查看消费者日志:
Received via @StreamListener: Hello Kafka Stream! // 或 Received via Function: Hello Kafka Stream!
常见问题
-
版本兼容性:
-
Spring Cloud Stream 3.x 后需使用函数式编程。
-
检查 Spring Boot 版本与 Spring Cloud Stream 的匹配关系(如 Spring Boot 2.6.x + Spring Cloud 2021.x)。
-
-
绑定配置:
-
函数式模型中,绑定名称格式为
<functionName>-in-<index>(如myInput-in-0)。
-
-
序列化配置:
-
若传递 JSON 对象,需配置
content-type=application/json并添加 Jackson 依赖。
-
总结
-
旧版:使用
@StreamListener+ 通道接口(适合遗留代码升级)。 -
新版:推荐函数式编程模型(更简洁,符合现代 Spring 设计)。
-
根据实际 Spring Cloud Stream 版本选择方案!
相关文章:
kafka+spring cloud stream 发送接收消息
方案 1:使用旧版 StreamListener(适用于 Spring Cloud Stream < 2.x) 1. 添加依赖(pom.xml) <!-- Spring Cloud Stream Kafka Binder --> <dependency> <groupId>org.springframework.clo…...
使用ArcGIS Pro自动矢量化水系
在地理信息系统(GIS)领域,自动矢量化是一项至关重要的技术,它能够将栅格图像中的要素转换为矢量数据,从而方便后续的分析和处理。本文将详细介绍如何使用ArcGIS Pro自动矢量化水系,适用于那些颜色相对统一、…...
在PyCharm中运行Jupyter Notebook的.ipynb文件及其pycharm软件的基础使用
(注意需使用PyCharm专业版,学生、教师可以申请免费使用:https://www.jetbrains.com/shop/eform/students) 1. pycharm2024版汉化 https://blog.csdn.net/m0_74103046/article/details/144560999 2. pycharm中的python控制台和J…...
实验 Figma MCP + Cursor 联合工作流
开源项目 Figma-Context-MCP 介绍 使用此 Model Context Protocol 服务器授予 Cursor 对 Figma 文件的访问权限。 当 Cursor 可以访问 Figma 设计数据时,它比粘贴屏幕截图等其他方法更能准确地进行代码转化。 开源仓库: GLips/Figma-Context-MCP 具体…...
移植live555 上的 rtsp
一、V4L2视频采集模块(完整示例) #include <linux/videodev2.h> #include <sys/ioctl.h> #include <fcntl.h>// 初始化V4L2摄像头 int init_v4l2_camera(const char* dev_path, int width, int height) {int fd open(dev_path, O_RD…...
Web Worker终极优化指南:4秒卡顿→0延迟的实战蜕变
💡 导读:从4秒卡顿到丝滑响应 真实痛点场景:当斐波那契数列计算量达10亿次时,页面完全冻结4.2秒!通过Web Worker优化后,UI响应时间降至16ms以内。本文手把手带您实现性能蜕变! 一、Web Worker核…...
redis中的Lua脚本,redis的事务机制
lua脚本的特点 lua脚本可以操作redis数据库,并且脚本中的代码满足原子性,要么全部被执行,要么全部不执行 lua脚本的语法 脚本示例 lua脚本的草稿: 最终的lua脚本 lua脚本在java里调用的方法 RedisTemplete类里有一个方法&…...
CPU多级缓存与缓存一致性协议
CPU多级缓存与缓存一致性协议 CPU多级缓存和缓存一致性协议是计算机体系结构中优化性能与保证数据正确性的核心机制。以下从缓存层级设计、工作原理、一致性协议(如MESI)及其实现细节展开说明。 一、为什么需要多级缓存? CPU的计算速度远高…...
Apifox 增强 AI 接口调试功能:自动合并 SSE 响应、展示DeepSeek思考过程
在现代的API接口调试中,效率和精确性对于开发者和测试人员来说至关重要。Apifox,作为一款功能强大的API管理和调试工具,近年来不断提升其用户体验和智能化功能。最近,Apifox 推出了增强版的AI接口调试功能,其中包括自动…...
【电机控制】42步进电机+arduino:WHEELTEC_MS42DDC
轮趣科技 42步进电机arduino:WHEELTEC_MS42DDC 接线方式: WHEELTEC_MS42DDC有两个接口, 一端接口连接配套的DC电源,另外一端只需要用三根线,一根负极连接ardino 的GND,然后把该端口的tx和rx连接到arduino的rx和tx,下…...
使用LangChain构建第一个ReAct Agent
使用LangChain构建第一个ReAct Agent 准备环境 使用Anaconda 安装python 3.10 安装langchain、langchain_openai、langchain_community (安装命令 pip install XXX) 申请DeepSeek API:https://platform.deepseek.com/api_keys(也…...
萝卜头笔作文赏析
在遥远的无寻王国,有这么一支小小的笔诞生了,人们见它又短又小,于是就给它取名叫萝卜头笔。萝卜头笔渐渐长大了,除了身子变粗些,其他什么都没变。一天,萝卜头笔来到了深山老林,那里枝叶繁茂&…...
RT-Thread+STM32L475VET6——USB鼠标模拟
文章目录 前言一、板载资源二、具体步骤1.配置icm20608传感器2.打开CubeMX进行USB配置3. 配置USB3.1 打开USB驱动3.2 声明USB3.3 剪切stm32xxxx_hal_msp.c中的void HAL_PCD_MspInit(PCD_HandleTypeDef* hpcd)和void HAL_PCD_MspDeInit(PCD_HandleTypeDef* hpcd)函数至board.c3.…...
rust 安全性
Rust 是 静态类型(statically typed) 语言, 也就是说在编译时就必须知道所有变量的类型, 这一点将贯穿整个章节。 C/C的安全问题 内存的不正确访问引发的内存安全问题 由于多个变量指向同一块内存区域导致的数据一致性问题 由于…...
大模型驱动的围术期质控系统全面解析与应用探索
目录 一、引言 1.1 研究背景与意义 1.2 研究目的与方法 1.3 研究创新点 二、大模型技术与围术期管理概述 2.1 大模型技术原理与发展现状 2.2 围术期管理流程与挑战 三、大模型在术前的应用 3.1 病历内涵质控 3.2 智能医学问答与知识查询 3.3 疾病风险预测与评估 3.…...
中兴B863AV3.2-T/B863AV3.1-T2/B863AV3.1-T2K_电信高安_S905L3A-B_安卓9.0_线刷固件包
中兴B863AV3.2-T/B863AV3.1-T2/B863AV3.1-T2K_电信高安_S905L3A-B_安卓9.0_线刷固件包 B863AV3.2-T B863AV3.1-T2 已知可通刷贵州、江苏、贵州、北京、河南、陕西等省份。 线刷方法:(新手参考借鉴一下) 1、准备好一…...
Android Binder机制
Binder是IPC(进程间通信)的一种机制,它允许不同的应用或系统服务在不同的进程中安全地交换数据。Binder的核心原理是基于客户端-服务器模型(C/S架构)。 一、Binder的定义 1. Binder是Android中的一个类,它继承了IBind…...
【算法】初等数论
初等数论 模 取余,遵循尽可能让商向0靠近的原则,结果的正负和左操作数相同 取模,遵循尽可能让商向负无穷靠近的原则,结果的正负和右操作数相同 7/(-3)-2.3,产生了两个商-2和-3,取…...
Spring Boot3+Vue2极速整合:10分钟搭建DeepSeek AI对话系统
前言 在生成式AI技术蓬勃发展的今天,大语言模型已成为企业智能化转型和个人效率提升的核心驱动力。作为国产大模型的优秀代表,DeepSeek凭借其卓越的中文语义理解能力和开发者友好的API生态,正在成为构建本土化AI应用的首选平台。 本文将以S…...
Spring事务原理 二
在上一篇博文《Spring事务原理 一》中,我们熟悉了Spring声明式事务的AOP原理,以及事务执行的大体流程。 本文中,介绍了Spring事务的核心组件、传播行为的源码实现。下一篇中,我们将结合案例,来讲解实战中有关事务的易…...
Linux链表操作全解析
Linux C语言链表深度解析与实战技巧 一、链表基础概念与内核链表优势1.1 为什么使用链表?1.2 Linux 内核链表与用户态链表的区别 二、内核链表结构与宏解析常用宏/函数 三、内核链表的优点四、用户态链表示例五、双向循环链表在内核中的实现优势5.1 插入效率5.2 安全…...
K8S认证|CKS题库+答案| 11. AppArmor
目录 11. AppArmor 免费获取并激活 CKA_v1.31_模拟系统 题目 开始操作: 1)、切换集群 2)、切换节点 3)、切换到 apparmor 的目录 4)、执行 apparmor 策略模块 5)、修改 pod 文件 6)、…...
在鸿蒙HarmonyOS 5中实现抖音风格的点赞功能
下面我将详细介绍如何使用HarmonyOS SDK在HarmonyOS 5中实现类似抖音的点赞功能,包括动画效果、数据同步和交互优化。 1. 基础点赞功能实现 1.1 创建数据模型 // VideoModel.ets export class VideoModel {id: string "";title: string ""…...
Cesium1.95中高性能加载1500个点
一、基本方式: 图标使用.png比.svg性能要好 <template><div id"cesiumContainer"></div><div class"toolbar"><button id"resetButton">重新生成点</button><span id"countDisplay&qu…...
基础测试工具使用经验
背景 vtune,perf, nsight system等基础测试工具,都是用过的,但是没有记录,都逐渐忘了。所以写这篇博客总结记录一下,只要以后发现新的用法,就记得来编辑补充一下 perf 比较基础的用法: 先改这…...
2025盘古石杯决赛【手机取证】
前言 第三届盘古石杯国际电子数据取证大赛决赛 最后一题没有解出来,实在找不到,希望有大佬教一下我。 还有就会议时间,我感觉不是图片时间,因为在电脑看到是其他时间用老会议系统开的会。 手机取证 1、分析鸿蒙手机检材&#x…...
自然语言处理——循环神经网络
自然语言处理——循环神经网络 循环神经网络应用到基于机器学习的自然语言处理任务序列到类别同步的序列到序列模式异步的序列到序列模式 参数学习和长程依赖问题基于门控的循环神经网络门控循环单元(GRU)长短期记忆神经网络(LSTM)…...
OPenCV CUDA模块图像处理-----对图像执行 均值漂移滤波(Mean Shift Filtering)函数meanShiftFiltering()
操作系统:ubuntu22.04 OpenCV版本:OpenCV4.9 IDE:Visual Studio Code 编程语言:C11 算法描述 在 GPU 上对图像执行 均值漂移滤波(Mean Shift Filtering),用于图像分割或平滑处理。 该函数将输入图像中的…...
网站指纹识别
网站指纹识别 网站的最基本组成:服务器(操作系统)、中间件(web容器)、脚本语言、数据厍 为什么要了解这些?举个例子:发现了一个文件读取漏洞,我们需要读/etc/passwd,如…...
保姆级教程:在无网络无显卡的Windows电脑的vscode本地部署deepseek
文章目录 1 前言2 部署流程2.1 准备工作2.2 Ollama2.2.1 使用有网络的电脑下载Ollama2.2.2 安装Ollama(有网络的电脑)2.2.3 安装Ollama(无网络的电脑)2.2.4 安装验证2.2.5 修改大模型安装位置2.2.6 下载Deepseek模型 2.3 将deepse…...
