RabbitMQ实践——使用WebFlux响应式方式实时返回队列中消息
大纲
- Pom.xml
- 监听队列
- 实时返回消息
- 测试
- 完整代码
- 工程代码
在之前的案例中,我们在管理后台收发消息都是通过短连接的形式。本文我们将探索对队列中消息的实时读取,并通过流式数据返回给客户端。
webflux是反应式Web框架,客户端可以通过一个长连接和服务端相连,后续服务端可以通过该连接持续给客户端发送消息。可以达到:发送一次,多次接收的效果。
Pom.xml
由于我们要使用Rabbitmq,所以要新增如下依赖
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-stream</artifactId></dependency>
webflux的依赖如下:
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId></dependency><dependency><groupId>io.projectreactor</groupId><artifactId>reactor-core</artifactId><version>3.6.7</version></dependency>
监听队列
下面代码会返回一个监听队列的Container
private SimpleMessageListenerContainer getListener(String queueName, MessageListener messageListener) {lock.lock();try {SimpleMessageListenerContainer listener = listeners.get(queueName);if (listener == null && messageListener != null) {listener = new SimpleMessageListenerContainer();listener.setConnectionFactory(connectionFactory);listener.setQueueNames(queueName);listener.setMessageListener(messageListener);listeners.put(queueName, listener);}return listener;} finally {lock.unlock();}}
实时返回消息
一旦消费者读取到消息,onMessage方法会被调用。然后Flux的消费者会将消息投递到流上。
public Flux<String> listen(String queueName) {return Flux.create(emitter -> {SimpleMessageListenerContainer container = getListener(queueName, new MessageListener() {@Overridepublic void onMessage(Message message) {String msg = new String(message.getBody());System.out.println("listen function Received message: " + msg);emitter.next(msg);}});container.start();});}
测试
由于OpenApi不能支持实时展现流式数据,所以我们采用Postman来测试。
发送请求后,该页面一直处于滚动状态。
在管理后台发送一条消息
可以看到Postman收到了该消息
然后在发一条,Postman又会收到一条
这样我们就完成了“请求一次,多次返回”的效果。
完整代码
需要注意的是,返回的格式需要标记为produces = “text/event-stream”。
// controller
package com.rabbitmq.consumer.controller;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;import com.rabbitmq.consumer.service.ConsumerService;import reactor.core.publisher.Flux;@RestController
@RequestMapping("/consumer")
public class ConsumerController {@Autowiredprivate ConsumerService comsumerService;@GetMapping(value = "/listen", produces = "text/event-stream")public Flux<String> listen(@RequestParam String queueName) {return comsumerService.listen(queueName);}
}
// service
package com.rabbitmq.consumer.service;import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import jakarta.annotation.PostConstruct;
import reactor.core.publisher.Flux;@Service
public class ConsumerService {@Autowiredprivate RabbitTemplate rabbitTemplate;private ConnectionFactory connectionFactory;private final ReentrantLock lock = new ReentrantLock();private Map<String, SimpleMessageListenerContainer> listeners = new java.util.HashMap<>();@PostConstructpublic void init() {connectionFactory = rabbitTemplate.getConnectionFactory();}public Flux<String> listen(String queueName) {return Flux.create(emitter -> {SimpleMessageListenerContainer container = getListener(queueName, new MessageListener() {@Overridepublic void onMessage(Message message) {String msg = new String(message.getBody());System.out.println("listen function Received message: " + msg);emitter.next(msg);}});container.start();});}private SimpleMessageListenerContainer getListener(String queueName, MessageListener messageListener) {lock.lock();try {SimpleMessageListenerContainer listener = listeners.get(queueName);if (listener == null && messageListener != null) {listener = new SimpleMessageListenerContainer();listener.setConnectionFactory(connectionFactory);listener.setQueueNames(queueName);listener.setMessageListener(messageListener);listeners.put(queueName, listener);}return listener;} finally {lock.unlock();}}
}
工程代码
https://github.com/f304646673/RabbitMQDemo/tree/main/consumer
相关文章:

RabbitMQ实践——使用WebFlux响应式方式实时返回队列中消息
大纲 Pom.xml监听队列实时返回消息测试完整代码工程代码 在之前的案例中,我们在管理后台收发消息都是通过短连接的形式。本文我们将探索对队列中消息的实时读取,并通过流式数据返回给客户端。 webflux是反应式Web框架,客户端可以通过一个长连…...

SpringBoot前后端传递数据时常用的JSON格式数据是什么?【讲解JSON概念、语法、以及Java对象互转】
SpringBoot前后端传递数据时常用的JSON格式数据是什么? JSON概念JSON语法JSON的两种结构:JSON字符串和Java对象互转:objectMapper.writeValueAsString(person);objectMapper.readValue(jsonStr,Person.class); 在SpringMVC框架中,…...

mysql学习——SQL中的DQL和DCL
SQL中的DQL和DCL DQL基本查询条件查询聚合函数分组查询排序查询分页查询 DCL管理用户权限控制 学习黑马MySQL课程,记录笔记,用于复习。 DQL DQL英文全称是Data Query Language(数据查询语言),数据查询语言,用来查询数据库中表的记…...

windows系统上nginx搭建文件共享
1、下载windows版nginx 下载地址 2、配置nginx 编辑nginx.conf配置文件 在http模块下添加这个参数 underscores_in_headers on;#修改location内容,共享哪个文件夹,就写哪个文件夹,最后一定要跟上/,否则无法访问 location / {…...

星闪指向遥控,做家电交互的破壁人
“面壁者罗辑,我是你的破壁人。” 科幻小说《三体》中,当人类的基础科学被三体人封锁,变得停步不前,人类启动了自救的面壁计划,通过一次又一次破壁,找到战胜三体人的办法。 现实中,有一点已经成…...

SpringBoot使用AutoConfigure实现依赖库自动导入配置
我们知道导入配置有两种,一种是Value,一种是ConfigurationProperties,将对应的类标记为Component即可导入。但是被注解标识的类创建Bean有一个前提,只对启动类所在的包路径下的所有带有Component等注解的类才会创建Bean。如果我们…...

QT中利用动画弄一个侧边栏窗口,以及贴条效果
1、效果 2、关键代码 void Widget::on_sliderBtn_clicked() {m_sliderWidget->show();QPropertyAnimation* animation = new QPropertyAnimation(m...

win10免安装配置MySQL8.4.0
注:此教程基于win10 22H2 版本 1、下载最新版本MySQL压缩包 下载链接:MySQL官网下载地址 点击第二行的 ZIP Archive 后面的Download(当前时间2024-06-19最新版本是8.4.0) 2、解压并添加配置文件 下载完毕后,解压缩…...

VS Code安装及环境配置(超详细)
VS Code简介 Visual Studio Code(简称 VS Code )是 Microsoft 于2015年4月发布的一款代码编辑器,以界面简洁、轻量著称。 它是一款免费开源的现代化轻量级代码编辑器,支持几乎所有主流开发语言的语法高亮、智能代码补全、自定义…...
shell脚本通过解析日志使用串口开关屏知识点整理
#!/bin/bash #logPath 写日志的存放路径 #logPath/home/workspace/tvs/trainborne logPath/home/firefly tmpFile$$ function getLogName() { #echo "$logPath/LCDController_"date "%Y%m%d000000.log" echo "LCDController_20240424000000.log&quo…...
速盾:视频cdn和网站cdn的相同点与不同点
CDN(Content Delivery Network)是一种分布式网络架构,旨在为用户提供高效、高质量的内容传送服务。CDN主要通过将内容分发到全球各地的边缘节点,并根据用户的地理位置选择最近的节点来提供内容,从而加速内容的传输并降…...
37.自定义协议
自定义协议要素 1.魔数:用来第一时间判定是否是无效数据包。 2.版本号:主持协议升级。 3.序列化算法:消息正文到底采用那种序列化方式,可以由此扩展,例如:json,protobuf(google的序列化算法,基于二进制,虽然可读性不好,但是字节数占用更少),hessian(基于二进制)…...

【React Native】measureInWindow在安卓上无法正确获取View在屏幕上的布局信息
问题描述: 在React Native中,我们可以使用measureInWindow的方式去获取一个View在屏幕中的位置信息: 下面这个Demo中,我们写了一个页面HomePage和一个列表项组件ListItemA,我们期望每过5s监测一次列表中每一项在屏幕中…...

C++ 教程 - 04 类的使用
文章目录 类的定义类定义案例构造函数 类的定义 C 在 C 语言的基础上增加面向对象编程,类是用于指定对象的形式,是一种用户自定义的数据类型,封装了数据和函数。类可以被看作是一种模板,可以用来创建具有相同属性和行为的多个对象…...
excel按模板文件导出多个文件并压缩为ZIP格式返回前端
思路:先准备好模板文件和与之对应的实体类,数据库数据等,还是之前思路,根据查出的数据,填充模板文件,生成一个临时文件,最后将这些个临时文件打包为zip返回前端,并将多个临时文件删除…...

自动驾驶仿真测试用例表格示例 ACC ELK FCW
自动驾驶仿真测试用例表格示例 测试用例概览 本测试用例表格涵盖了自动驾驶系统中多个关键功能和场景的测试,旨在确保系统在不同条件下的表现和稳定性。 用例编号测试项目测试描述预期结果实际结果通过/失败TC-001ACC功能测试在高速公路上启用ACC,测试车…...

数组 (java)
文章目录 一维数组静态初始化动态初始化 二维数组静态初始化动态初始化 数组参数传递可变参数关于 main 方法的形参 argsArray 工具类sort 中的 comparable 和 comparatorcomparator 比较器排序comparable 自然排序 一维数组 线性结构 静态初始化 第一种:int[] a…...

时序预测 | Matlab基于Transformer多变量时间序列多步预测
目录 效果一览基本介绍程序设计参考资料 效果一览 基本介绍 1.Matlab基于Transformer多变量时间序列多步预测; 2.多变量时间序列数据集(负荷数据集),采用前96个时刻预测的特征和负荷数据预测未来96个时刻的负荷数据; 3…...

suuk-s.php.jpg-python 库劫持
做virtualBox的端口映射吧 suukmedim文件白名单绕过、反弹shell、$paht环境变量更改、python 库劫持提权、Reptile提权、sandfly-processdecloak使用 服务扫描 ┌──(kali㉿kali)-[~] └─$ sudo nmap -sV -A -T 4 -p 22,80 192.168.18.238GetSHell 访问80http://192.168.…...

python3GUI--ktv点歌软件By:PyQt5(附下载地址)
文章目录 一.前言二.展示1.启动2.搜索2.服务1.首页2.天气预报3.酒水饮料4.酒水饮料2 3.服务4.灯光5.调音6.排行榜7.分类点歌9.歌手点歌10.歌手个人页 三.心得体会1.关于代码2.关于设计3.关于打包 四.总结 文件大小:33.…...

【kafka】Golang实现分布式Masscan任务调度系统
要求: 输出两个程序,一个命令行程序(命令行参数用flag)和一个服务端程序。 命令行程序支持通过命令行参数配置下发IP或IP段、端口、扫描带宽,然后将消息推送到kafka里面。 服务端程序: 从kafka消费者接收…...
五年级数学知识边界总结思考-下册
目录 一、背景二、过程1.观察物体小学五年级下册“观察物体”知识点详解:由来、作用与意义**一、知识点核心内容****二、知识点的由来:从生活实践到数学抽象****三、知识的作用:解决实际问题的工具****四、学习的意义:培养核心素养…...

(二)原型模式
原型的功能是将一个已经存在的对象作为源目标,其余对象都是通过这个源目标创建。发挥复制的作用就是原型模式的核心思想。 一、源型模式的定义 原型模式是指第二次创建对象可以通过复制已经存在的原型对象来实现,忽略对象创建过程中的其它细节。 📌 核心特点: 避免重复初…...
工业自动化时代的精准装配革新:迁移科技3D视觉系统如何重塑机器人定位装配
AI3D视觉的工业赋能者 迁移科技成立于2017年,作为行业领先的3D工业相机及视觉系统供应商,累计完成数亿元融资。其核心技术覆盖硬件设计、算法优化及软件集成,通过稳定、易用、高回报的AI3D视觉系统,为汽车、新能源、金属制造等行…...
OpenPrompt 和直接对提示词的嵌入向量进行训练有什么区别
OpenPrompt 和直接对提示词的嵌入向量进行训练有什么区别 直接训练提示词嵌入向量的核心区别 您提到的代码: prompt_embedding = initial_embedding.clone().requires_grad_(True) optimizer = torch.optim.Adam([prompt_embedding...
爬虫基础学习day2
# 爬虫设计领域 工商:企查查、天眼查短视频:抖音、快手、西瓜 ---> 飞瓜电商:京东、淘宝、聚美优品、亚马逊 ---> 分析店铺经营决策标题、排名航空:抓取所有航空公司价格 ---> 去哪儿自媒体:采集自媒体数据进…...

听写流程自动化实践,轻量级教育辅助
随着智能教育工具的发展,越来越多的传统学习方式正在被数字化、自动化所优化。听写作为语文、英语等学科中重要的基础训练形式,也迎来了更高效的解决方案。 这是一款轻量但功能强大的听写辅助工具。它是基于本地词库与可选在线语音引擎构建,…...
C++.OpenGL (20/64)混合(Blending)
混合(Blending) 透明效果核心原理 #mermaid-svg-SWG0UzVfJms7Sm3e {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-SWG0UzVfJms7Sm3e .error-icon{fill:#552222;}#mermaid-svg-SWG0UzVfJms7Sm3e .error-text{fill…...

RabbitMQ入门4.1.0版本(基于java、SpringBoot操作)
RabbitMQ 一、RabbitMQ概述 RabbitMQ RabbitMQ最初由LShift和CohesiveFT于2007年开发,后来由Pivotal Software Inc.(现为VMware子公司)接管。RabbitMQ 是一个开源的消息代理和队列服务器,用 Erlang 语言编写。广泛应用于各种分布…...
PostgreSQL——环境搭建
一、Linux # 安装 PostgreSQL 15 仓库 sudo dnf install -y https://download.postgresql.org/pub/repos/yum/reporpms/EL-$(rpm -E %{rhel})-x86_64/pgdg-redhat-repo-latest.noarch.rpm# 安装之前先确认是否已经存在PostgreSQL rpm -qa | grep postgres# 如果存在࿰…...