Rabbit SpringBoot高级用法
Rabbit高级用法
- 一、Rabbit Springboot集成
- 1.1 引入依赖
- 1.2 添加配置
- 1.3 添加Config
- 1.4 编写Consumer
- 1.5 发送消息
- 二、Rabbit 高级用法
- 2.1 消息发送前置处理器
- 2.2 消息发送确认机制
- 2.3 消息接收后处理器
- 2.4 事务消息
一、Rabbit Springboot集成
1.1 引入依赖
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1.2 添加配置
server:port: 8080
spring:application:name: rabbitmq-testrabbitmq:host: 127.0.0.1port: 5672username: testpassword: testvirtual-host: /publisher-returns: true
1.3 添加Config
@Configuration
@EnableConfigurationProperties(MqProperties.class)
public class MqConfig {@Autowiredprivate MqProperties mqProperties;@Beanpublic MessageConverter messageConverter() {// 设置消息转换器return new Jackson2JsonMessageConverter();}@Beanpublic RabbitTemplate rabbitTemplate() {RabbitTemplate rabbitTemplate = new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory());// 设置消息转换器rabbitTemplate.setMessageConverter(messageConverter());return rabbitTemplate;}@Beanpublic ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory();connectionFactory.setHost(mqProperties.getHost());connectionFactory.setPort(mqProperties.getPort());connectionFactory.setUsername(mqProperties.getUsername());connectionFactory.setPassword(mqProperties.getPassword());connectionFactory.setVirtualHost(mqProperties.getVirtualHost());connectionFactory.setPublisherReturns(mqProperties.getPublisherReturns());connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);return connectionFactory;}@Beanpublic Queue queue() {return new Queue("test-queue", true, false, false);}@Beanpublic Exchange exchange() {return new DirectExchange("direct-exchange-test", true, false);}@Beanpublic Binding binding() {return BindingBuilder.bind(queue()).to(exchange()).with("direct-key").noargs();}
}
1.4 编写Consumer
@Component
public class DirectConsumer extends MessageListenerAdapter {private static final Logger logger = LoggerFactory.getLogger(DirectConsumer.class);@Autowiredprivate MessageConverter messageConverter;@Override@RabbitListener(queues = {"test-queue"}, ackMode = "MANUAL")public void onMessage(Message message, Channel channel) throws Exception {try {Map<String, String> msg = (Map<String, String>) messageConverter.fromMessage(message);// 获取 correlation idString id = (String) message.getMessageProperties().getHeaders().get(PublisherCallbackChannel.RETURNED_MESSAGE_CORRELATION_KEY);// String msg = new String(message.getBody(), StandardCharsets.UTF_8);logger.info("directConsumer>>>>>>message={}", msg);channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {logger.error("directConsumer>>>>>>exception", e);channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}}
}
1.5 发送消息
@RestController
@RequestMapping("/mq")
public class MqController {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/sendMsg/{data}")public void test(@PathVariable(value = "data", required = false) String data) {Map<String, String> msg = new HashMap<>();msg.put("time", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));if (StringUtils.isEmpty(data)) {data = String.valueOf(System.currentTimeMillis());}msg.put("data", "this is data:" + data);rabbitTemplate.convertAndSend("direct-exchange-test", "direct-key", msg, new CorrelationData(UUID.randomUUID().toString()));}
}
二、Rabbit 高级用法
2.1 消息发送前置处理器
RabbitTemplate.setBeforePublishPostProcessors(new MessagePostProcessor() {....})
@Bean
public RabbitTemplate rabbitTemplate() {RabbitTemplate rabbitTemplate = new RabbitTemplate();// 添加消息前置处理器。rabbitTemplate.setBeforePublishPostProcessors(new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {logger.info("-----------------rabbitmq before postProcess message={}", JSON.toJSONString(message));return message;}});rabbitTemplate.setConnectionFactory(connectionFactory());// 设置消息转换器rabbitTemplate.setMessageConverter(messageConverter());return rabbitTemplate;
}
通过前置处理器,可以修改消息、保存消息,设置通用header等。
2.2 消息发送确认机制
设置消息发送ConfirmCallback,消息发送成功 / 失败都会调用当前方法。
@Bean
public RabbitTemplate rabbitTemplate() {RabbitTemplate rabbitTemplate = new RabbitTemplate();rabbitTemplate.setCorrelationDataPostProcessor(new CorrelationDataPostProcessor() {@Overridepublic CorrelationData postProcess(Message message, CorrelationData correlationData) {logger.info("-----------------rabbitmq correlationData postProcess message={}, correlationData={}",JSON.toJSONString(message), JSON.toJSONString(correlationData));return correlationData;}});rabbitTemplate.setConnectionFactory(connectionFactory());// 设置消息转换器rabbitTemplate.setMessageConverter(messageConverter());// 消息确认,需要配置 spring.rabbitmq.publisher-confirm-type = correlatedrabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {logger.info("setConfirmCallback>>>>>>correlationData={} ack={}, cause={}", JSON.toJSONString(correlationData), ack, cause);}});
//开启mandatory模式(开启失败回调)rabbitTemplate.setMandatory(true);//添加失败回调方法rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {logger.info("setReturnCallback>>>>>消息发送队列不可达, message:{}, exchange:{}, routingKey:{}, 原因:{}", message, exchange, routingKey, replyText);});return rabbitTemplate;
}
通过 new RabbitTemplate.ConfirmCallback() 中的 confirm(CorrelationData correlationData, boolean ack, String cause) 判断消息是否发送成功。
- ack=true:发送成功
- ack=false:发送失败
2.3 消息接收后处理器
消息接收前执行。
方式1:
/**
* 添加SimpleRabbitListenerContainerFactory
* 通过 @RabbitListener(queues = {"test-queue"}, containerFactory = "containerFactory", ackMode = "MANUAL") 设置
*/
@Bean("containerFactory")
public SimpleRabbitListenerContainerFactory containerFactory(MessageConverter messageConverter) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();// 并发消费者数,默认为 1factory.setConcurrentConsumers(5);// 最大并发消费者数,默认为 1factory.setMaxConcurrentConsumers(10);// 拒绝未确认的消息并重新将它们放回队列,默认为 truefactory.setDefaultRequeueRejected(false);// 容器启动时是否自动启动,默认为 truefactory.setAutoStartup(true);// 消息确认模式,默认为 AUTOfactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 每个消费者在一次请求中预获取的消息数,默认为 1factory.setPrefetchCount(5);// 从队列中接收消息的超时时间,默认为 0,表示没有超时限制factory.setReceiveTimeout(0L);// 与容器一起使用的事务管理器。默认情况下,容器不会使用事务// factory.setTransactionManager(platformTransactionManager());// 消息转换器,用于将接收到的消息转换为 Java 对象或将 Java 对象转换为消息factory.setMessageConverter(messageConverter);// 用于异步消息处理的线程池。默认情况下,容器使用一个简单的 SimpleAsyncTaskExecutorfactory.setTaskExecutor(new SimpleAsyncTaskExecutor());// 重试失败的消息之前等待的时间,默认为 5000 毫秒factory.setRecoveryInterval(5000L);// 如果消息处理器尝试监听不存在的队列,是否抛出异常。默认为 truefactory.setMissingQueuesFatal(false);// 监听器容器连接工厂factory.setConnectionFactory(connectionFactory());// 设置后置处理器factory.setAfterReceivePostProcessors(new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {logger.info("-----------------rabbitmq after postProcess message={}", JSON.toJSONString(message));return message;}});return factory;
}
方式2:
@Bean
public SimpleMessageListenerContainer listenerContainer(@Qualifier("directConsumer") DirectConsumer directConsumer) {String queueName = "test-queue";SimpleMessageListenerContainer simpleMessageListenerContainer =new SimpleMessageListenerContainer(connectionFactory());simpleMessageListenerContainer.setQueueNames(queueName);simpleMessageListenerContainer.setMessageListener(directConsumer);return simpleMessageListenerContainer;
}@Autowired(required = false)
private List<AbstractMessageListenerContainer> simpleMessageListenerContainers;@PostConstruct
public void init() {if (CollectionUtils.isEmpty(simpleMessageListenerContainers)) {return;}for (AbstractMessageListenerContainer simpleMessageListenerContainer : simpleMessageListenerContainers) {simpleMessageListenerContainer.setAfterReceivePostProcessors(new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {logger.info("-----------------rabbitmq after postProcess message={}", JSON.toJSONString(message));return message;}});// 设置手动 ACKsimpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);}
}
2.4 事务消息
不建议使用 rabbitmq 事务消息,对性能非常影响。建议通过消息发送确认机制实现事务。
相关文章:
Rabbit SpringBoot高级用法
Rabbit高级用法 一、Rabbit Springboot集成1.1 引入依赖1.2 添加配置1.3 添加Config1.4 编写Consumer1.5 发送消息 二、Rabbit 高级用法2.1 消息发送前置处理器2.2 消息发送确认机制2.3 消息接收后处理器2.4 事务消息 一、Rabbit Springboot集成 1.1 引入依赖 <dependency…...
找不到vcruntime140.dll,无法继续执行代码?多种解决方法解析
找不到vcruntime140.dll,无法继续执行代码?当你在尝试运行某个程序时,突然弹出一条错误提示框,告诉你无法继续执行代码,因为找不到vcruntime140.dll。这个问题很常见,但是它可能会让你感到困惑和疑惑。这篇文章将详细介…...
自然语言处理实战项目8- BERT模型的搭建,训练BERT实现实体抽取识别的任务
大家好,我是微学AI,今天给大家介绍一下自然语言处理实战项目8- BERT模型的搭建,训练BERT实现实体抽取识别的任务。BERT模型是一种用于自然语言处理的深度学习模型,它可以通过训练来理解单词之间的上下文关系,从而为下游…...
pdf怎么合并在一起?软件操作更高效
PDF格式已经成为了许多文档和表格的首选格式。然而,当你需要合并多个PDF文件时,可能会遇到一些麻烦,在本篇文章中,我们将向您介绍一种简单易用的方法来合并PDF文件。 以下是可以用来合并PDF文件的软件: - PDF转换器&a…...
Junit常见用法
一.Junit的含义 Junit是一种Java编程语言的单元测试框架。它提供了一些用于编写和运行测试的注释和断言方法,并且可以方便地执行测试并生成测试报告。Junit是开源的,也是广泛使用的单元测试框架之一 二.Junit项目的创建 (1)先创…...
c++—内存管理、智能指针、内存池
1. 内存分析诊断工具:valgrind; 2. 内存管理的两种方式: ①用户管理:自己申请的,自己用,自己回收;效率高,但容易导致内存泄漏; ②系统管理:系统自动回收垃圾…...
JAVA使用HTTP代码示例
以下是使用Java发送HTTP请求的示例代码: java import java.io.BufferedReader; import java.io.InputStreamReader; import java.net.HttpURLConnection; import java.net.URL; public class HttpExample { public static void main(String[] args) { try { …...
【网络协议详解】——电子邮件系统协议(学习笔记)
目录 🕒 1. 电子邮件系统概述🕒 2. 简单邮件传送协议SMTP🕒 3. SMTP协议的命令和响应🕘 3.1 命令🕤 3.1.1 HELO🕤 3.1.2 MAIL FROM🕤 3.1.3 RCPT TO🕤 3.1.4 DATA🕤 3.1.…...
年度发布 | MeterSphere一站式开源持续测试平台发布v2.10 LTS版本
2023年5月25日,MeterSphere一站式开源持续测试平台正式发布v2.10 LTS版本。这是继2022年5月发布v1.20 LTS版本后,MeterSphere开源项目发布的第三个LTS(Long Term Support)版本。MeterSphere开源项目组将对MeterSphere v2.10 LTS版…...
从 OceanBase 迁移数据到 DolphinDB
OceanBase 是一款金融级分布式关系数据库,具有数据强一致、高可用、高性能、在线扩展、高度兼容 SQL标准和主流关系数据库、低成本等特点,但是其学习成本较高,且缺乏金融计算函数以及流式增量计算的功能。 DolphinDB 是一款国产的高性能分布…...
淘宝商品列表数据接口(支持价格、销量排序)
淘宝商品列表数据接口是淘宝提供的一种可以获取淘宝商品信息的接口。通过该接口,可以获取到具有一定规则的商品信息,例如按照价格排序、按照销量排序等。接口返回的数据格式为JSON格式,可以方便地处理数据。 我们可以通过调用淘宝提供的API&…...
Android 11 版本变更总览
Android 11 版本 Android 11 总览重大隐私权变更行为变更:所有应用行为变更:以 Android 11 为目标平台的应用功能和 API 概览Intent系统广播 intent(API 级别 30)通用应用 intent(API 级别 30) Android 11 …...
传染病学模型 | Matlab实现基于SIS传染病模型模拟城市内人口的互相感染及城市人口流动所造成的传染
文章目录 效果一览基本描述模型介绍程序设计参考资料效果一览 基本描述 传染病学模型 | Matlab实现基于SIS传染病模型模拟城市内人口的互相感染及城市人口流动所造成的传染 模型介绍 SIS模型是一种基本的传染病学模型,用于描述一个人群中某种传染病的传播情况。SIS模型假设每个…...
物联网技术如何改变我们的生活:一位资深物联网专家的见解
物联网(IoT)是指通过网络互联的物理设备、车辆、建筑物以及其他物品,这些物品都内置了传感器、执行器、软件和网络连接器,使它们能够收集和交换数据。物联网技术已经在各个领域产生了深远的影响,包括家庭、医疗、交通、…...
node.js+vue.js大学生在线选课系统的设计与实现93pul
本次设计任务是要设计一个选课系统的设计与实现,通过这个系统能够满足用户对选课信息的需求。系统的主要功能包括:个人中心、学生管理、教师管理、选课信息管理等功能。 管理员可以根据系统给定的账号进行登录,登录后可以进入选课系统的设计与…...
华为OD机试真题 Java 实现【寻找符合要求的最长子串】【2023Q1 200分】
一、题目描述 给定一个字符串 s ,找出这样一个子串: 该子串中的任意一个字符最多出现2次;该子串不包含指定某个字符;请你找出满足该条件的最长子串的长度。 二、输入描述 第一行为要求不包含的指定字符,为单个字符,取值范围[0-9a-zA-Z]。 第二行为字符串s,每个字符范…...
接口测试工具Postman接口测试图文教程
目录 一、前言 二、Postman安装和使用 三、请求方式 四、资金记录接口实例演示 一、前言 在前后端分离开发时,后端工作人员完成系统接口开发后,需要与前端人员对接,测试调试接口,验证接口的正确性可用性。而这要求前端开发进…...
视频编辑软件:迅捷视频工具箱
这是一款功能强大、易于使用的视频编辑工具,支持视频剪辑、视频转换、音频转换、视频压缩、视频水印、字幕贴图等实用功能,可以帮助你制作出高质量的视频作品。(传送门:https://www.xunjiepdf.com/xjspgjx) 功能简介 …...
网络知识点之-HTTP协议
超文本传输协议(Hyper Text Transfer Protocol,HTTP)是一个简单的请求-响应协议,它通常运行在TCP之上。它指定了客户端可能发送给服务器什么样的消息以及得到什么样的响应。请求和响应消息的头以ASCII形式给出;而消息内…...
K类函数和KL类函数
Class K \mathcal{K} K function- K \mathcal{K} K类函数 Definition: A continuous function α : [ 0 , a ) → [ 0 , ∞ ) \alpha:[0,a)\rightarrow[0,\infin) α:[0,a)→[0,∞) is said belong to class K \mathcal{K} K if it strictly increasing and α ( 0 ) 0 …...
后进先出(LIFO)详解
LIFO 是 Last In, First Out 的缩写,中文译为后进先出。这是一种数据结构的工作原则,类似于一摞盘子或一叠书本: 最后放进去的元素最先出来 -想象往筒状容器里放盘子: (1)你放进的最后一个盘子(…...
IDEA运行Tomcat出现乱码问题解决汇总
最近正值期末周,有很多同学在写期末Java web作业时,运行tomcat出现乱码问题,经过多次解决与研究,我做了如下整理: 原因: IDEA本身编码与tomcat的编码与Windows编码不同导致,Windows 系统控制台…...
Vim 调用外部命令学习笔记
Vim 外部命令集成完全指南 文章目录 Vim 外部命令集成完全指南核心概念理解命令语法解析语法对比 常用外部命令详解文本排序与去重文本筛选与搜索高级 grep 搜索技巧文本替换与编辑字符处理高级文本处理编程语言处理其他实用命令 范围操作示例指定行范围处理复合命令示例 实用技…...
XCTF-web-easyupload
试了试php,php7,pht,phtml等,都没有用 尝试.user.ini 抓包修改将.user.ini修改为jpg图片 在上传一个123.jpg 用蚁剑连接,得到flag...
装饰模式(Decorator Pattern)重构java邮件发奖系统实战
前言 现在我们有个如下的需求,设计一个邮件发奖的小系统, 需求 1.数据验证 → 2. 敏感信息加密 → 3. 日志记录 → 4. 实际发送邮件 装饰器模式(Decorator Pattern)允许向一个现有的对象添加新的功能,同时又不改变其…...
RocketMQ延迟消息机制
两种延迟消息 RocketMQ中提供了两种延迟消息机制 指定固定的延迟级别 通过在Message中设定一个MessageDelayLevel参数,对应18个预设的延迟级别指定时间点的延迟级别 通过在Message中设定一个DeliverTimeMS指定一个Long类型表示的具体时间点。到了时间点后…...
Day131 | 灵神 | 回溯算法 | 子集型 子集
Day131 | 灵神 | 回溯算法 | 子集型 子集 78.子集 78. 子集 - 力扣(LeetCode) 思路: 笔者写过很多次这道题了,不想写题解了,大家看灵神讲解吧 回溯算法套路①子集型回溯【基础算法精讲 14】_哔哩哔哩_bilibili 完…...
【网络安全产品大调研系列】2. 体验漏洞扫描
前言 2023 年漏洞扫描服务市场规模预计为 3.06(十亿美元)。漏洞扫描服务市场行业预计将从 2024 年的 3.48(十亿美元)增长到 2032 年的 9.54(十亿美元)。预测期内漏洞扫描服务市场 CAGR(增长率&…...
UDP(Echoserver)
网络命令 Ping 命令 检测网络是否连通 使用方法: ping -c 次数 网址ping -c 3 www.baidu.comnetstat 命令 netstat 是一个用来查看网络状态的重要工具. 语法:netstat [选项] 功能:查看网络状态 常用选项: n 拒绝显示别名&#…...
CentOS下的分布式内存计算Spark环境部署
一、Spark 核心架构与应用场景 1.1 分布式计算引擎的核心优势 Spark 是基于内存的分布式计算框架,相比 MapReduce 具有以下核心优势: 内存计算:数据可常驻内存,迭代计算性能提升 10-100 倍(文档段落:3-79…...
