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

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,无法继续执行代码&#xff1f;当你在尝试运行某个程序时&#xff0c;突然弹出一条错误提示框&#xff0c;告诉你无法继续执行代码&#xff0c;因为找不到vcruntime140.dll。这个问题很常见&#xff0c;但是它可能会让你感到困惑和疑惑。这篇文章将详细介…...

自然语言处理实战项目8- BERT模型的搭建,训练BERT实现实体抽取识别的任务

大家好&#xff0c;我是微学AI&#xff0c;今天给大家介绍一下自然语言处理实战项目8- BERT模型的搭建&#xff0c;训练BERT实现实体抽取识别的任务。BERT模型是一种用于自然语言处理的深度学习模型&#xff0c;它可以通过训练来理解单词之间的上下文关系&#xff0c;从而为下游…...

pdf怎么合并在一起?软件操作更高效

PDF格式已经成为了许多文档和表格的首选格式。然而&#xff0c;当你需要合并多个PDF文件时&#xff0c;可能会遇到一些麻烦&#xff0c;在本篇文章中&#xff0c;我们将向您介绍一种简单易用的方法来合并PDF文件。 以下是可以用来合并PDF文件的软件&#xff1a; - PDF转换器&a…...

Junit常见用法

一.Junit的含义 Junit是一种Java编程语言的单元测试框架。它提供了一些用于编写和运行测试的注释和断言方法&#xff0c;并且可以方便地执行测试并生成测试报告。Junit是开源的&#xff0c;也是广泛使用的单元测试框架之一 二.Junit项目的创建 &#xff08;1&#xff09;先创…...

c++—内存管理、智能指针、内存池

1. 内存分析诊断工具&#xff1a;valgrind&#xff1b; 2. 内存管理的两种方式&#xff1a; ①用户管理&#xff1a;自己申请的&#xff0c;自己用&#xff0c;自己回收&#xff1b;效率高&#xff0c;但容易导致内存泄漏&#xff1b; ②系统管理&#xff1a;系统自动回收垃圾…...

JAVA使用HTTP代码示例

以下是使用Java发送HTTP请求的示例代码&#xff1a; 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 { …...

【网络协议详解】——电子邮件系统协议(学习笔记)

目录 &#x1f552; 1. 电子邮件系统概述&#x1f552; 2. 简单邮件传送协议SMTP&#x1f552; 3. SMTP协议的命令和响应&#x1f558; 3.1 命令&#x1f564; 3.1.1 HELO&#x1f564; 3.1.2 MAIL FROM&#x1f564; 3.1.3 RCPT TO&#x1f564; 3.1.4 DATA&#x1f564; 3.1.…...

年度发布 | MeterSphere一站式开源持续测试平台发布v2.10 LTS版本

2023年5月25日&#xff0c;MeterSphere一站式开源持续测试平台正式发布v2.10 LTS版本。这是继2022年5月发布v1.20 LTS版本后&#xff0c;MeterSphere开源项目发布的第三个LTS&#xff08;Long Term Support&#xff09;版本。MeterSphere开源项目组将对MeterSphere v2.10 LTS版…...

从 OceanBase 迁移数据到 DolphinDB

OceanBase 是一款金融级分布式关系数据库&#xff0c;具有数据强一致、高可用、高性能、在线扩展、高度兼容 SQL标准和主流关系数据库、低成本等特点&#xff0c;但是其学习成本较高&#xff0c;且缺乏金融计算函数以及流式增量计算的功能。 DolphinDB 是一款国产的高性能分布…...

淘宝商品列表数据接口(支持价格、销量排序)

淘宝商品列表数据接口是淘宝提供的一种可以获取淘宝商品信息的接口。通过该接口&#xff0c;可以获取到具有一定规则的商品信息&#xff0c;例如按照价格排序、按照销量排序等。接口返回的数据格式为JSON格式&#xff0c;可以方便地处理数据。 我们可以通过调用淘宝提供的API&…...

Android 11 版本变更总览

Android 11 版本 Android 11 总览重大隐私权变更行为变更&#xff1a;所有应用行为变更&#xff1a;以 Android 11 为目标平台的应用功能和 API 概览Intent系统广播 intent&#xff08;API 级别 30&#xff09;通用应用 intent&#xff08;API 级别 30&#xff09; Android 11 …...

传染病学模型 | Matlab实现基于SIS传染病模型模拟城市内人口的互相感染及城市人口流动所造成的传染

文章目录 效果一览基本描述模型介绍程序设计参考资料效果一览 基本描述 传染病学模型 | Matlab实现基于SIS传染病模型模拟城市内人口的互相感染及城市人口流动所造成的传染 模型介绍 SIS模型是一种基本的传染病学模型,用于描述一个人群中某种传染病的传播情况。SIS模型假设每个…...

物联网技术如何改变我们的生活:一位资深物联网专家的见解

物联网&#xff08;IoT&#xff09;是指通过网络互联的物理设备、车辆、建筑物以及其他物品&#xff0c;这些物品都内置了传感器、执行器、软件和网络连接器&#xff0c;使它们能够收集和交换数据。物联网技术已经在各个领域产生了深远的影响&#xff0c;包括家庭、医疗、交通、…...

node.js+vue.js大学生在线选课系统的设计与实现93pul

本次设计任务是要设计一个选课系统的设计与实现&#xff0c;通过这个系统能够满足用户对选课信息的需求。系统的主要功能包括&#xff1a;个人中心、学生管理、教师管理、选课信息管理等功能。 管理员可以根据系统给定的账号进行登录&#xff0c;登录后可以进入选课系统的设计与…...

华为OD机试真题 Java 实现【寻找符合要求的最长子串】【2023Q1 200分】

一、题目描述 给定一个字符串 s ,找出这样一个子串: 该子串中的任意一个字符最多出现2次;该子串不包含指定某个字符;请你找出满足该条件的最长子串的长度。 二、输入描述 第一行为要求不包含的指定字符,为单个字符,取值范围[0-9a-zA-Z]。 第二行为字符串s,每个字符范…...

接口测试工具Postman接口测试图文教程

目录 一、前言 二、Postman安装和使用 三、请求方式 四、资金记录接口实例演示 一、前言 在前后端分离开发时&#xff0c;后端工作人员完成系统接口开发后&#xff0c;需要与前端人员对接&#xff0c;测试调试接口&#xff0c;验证接口的正确性可用性。而这要求前端开发进…...

视频编辑软件:迅捷视频工具箱

这是一款功能强大、易于使用的视频编辑工具&#xff0c;支持视频剪辑、视频转换、音频转换、视频压缩、视频水印、字幕贴图等实用功能&#xff0c;可以帮助你制作出高质量的视频作品。&#xff08;传送门&#xff1a;https://www.xunjiepdf.com/xjspgjx&#xff09; 功能简介 …...

网络知识点之-HTTP协议

超文本传输协议&#xff08;Hyper Text Transfer Protocol&#xff0c;HTTP&#xff09;是一个简单的请求-响应协议&#xff0c;它通常运行在TCP之上。它指定了客户端可能发送给服务器什么样的消息以及得到什么样的响应。请求和响应消息的头以ASCII形式给出&#xff1b;而消息内…...

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 …...

云原生核心技术 (7/12): K8s 核心概念白话解读(上):Pod 和 Deployment 究竟是什么?

大家好&#xff0c;欢迎来到《云原生核心技术》系列的第七篇&#xff01; 在上一篇&#xff0c;我们成功地使用 Minikube 或 kind 在自己的电脑上搭建起了一个迷你但功能完备的 Kubernetes 集群。现在&#xff0c;我们就像一个拥有了一块崭新数字土地的农场主&#xff0c;是时…...

突破不可导策略的训练难题:零阶优化与强化学习的深度嵌合

强化学习&#xff08;Reinforcement Learning, RL&#xff09;是工业领域智能控制的重要方法。它的基本原理是将最优控制问题建模为马尔可夫决策过程&#xff0c;然后使用强化学习的Actor-Critic机制&#xff08;中文译作“知行互动”机制&#xff09;&#xff0c;逐步迭代求解…...

QMC5883L的驱动

简介 本篇文章的代码已经上传到了github上面&#xff0c;开源代码 作为一个电子罗盘模块&#xff0c;我们可以通过I2C从中获取偏航角yaw&#xff0c;相对于六轴陀螺仪的yaw&#xff0c;qmc5883l几乎不会零飘并且成本较低。 参考资料 QMC5883L磁场传感器驱动 QMC5883L磁力计…...

linux 下常用变更-8

1、删除普通用户 查询用户初始UID和GIDls -l /home/ ###家目录中查看UID cat /etc/group ###此文件查看GID删除用户1.编辑文件 /etc/passwd 找到对应的行&#xff0c;YW343:x:0:0::/home/YW343:/bin/bash 2.将标红的位置修改为用户对应初始UID和GID&#xff1a; YW3…...

如何理解 IP 数据报中的 TTL?

目录 前言理解 前言 面试灵魂一问&#xff1a;说说对 IP 数据报中 TTL 的理解&#xff1f;我们都知道&#xff0c;IP 数据报由首部和数据两部分组成&#xff0c;首部又分为两部分&#xff1a;固定部分和可变部分&#xff0c;共占 20 字节&#xff0c;而即将讨论的 TTL 就位于首…...

【HarmonyOS 5 开发速记】如何获取用户信息(头像/昵称/手机号)

1.获取 authorizationCode&#xff1a; 2.利用 authorizationCode 获取 accessToken&#xff1a;文档中心 3.获取手机&#xff1a;文档中心 4.获取昵称头像&#xff1a;文档中心 首先创建 request 若要获取手机号&#xff0c;scope必填 phone&#xff0c;permissions 必填 …...

【Oracle】分区表

个人主页&#xff1a;Guiat 归属专栏&#xff1a;Oracle 文章目录 1. 分区表基础概述1.1 分区表的概念与优势1.2 分区类型概览1.3 分区表的工作原理 2. 范围分区 (RANGE Partitioning)2.1 基础范围分区2.1.1 按日期范围分区2.1.2 按数值范围分区 2.2 间隔分区 (INTERVAL Partit…...

Web中间件--tomcat学习

Web中间件–tomcat Java虚拟机详解 什么是JAVA虚拟机 Java虚拟机是一个抽象的计算机&#xff0c;它可以执行Java字节码。Java虚拟机是Java平台的一部分&#xff0c;Java平台由Java语言、Java API和Java虚拟机组成。Java虚拟机的主要作用是将Java字节码转换为机器代码&#x…...

MySQL 索引底层结构揭秘:B-Tree 与 B+Tree 的区别与应用

文章目录 一、背景知识&#xff1a;什么是 B-Tree 和 BTree&#xff1f; B-Tree&#xff08;平衡多路查找树&#xff09; BTree&#xff08;B-Tree 的变种&#xff09; 二、结构对比&#xff1a;一张图看懂 三、为什么 MySQL InnoDB 选择 BTree&#xff1f; 1. 范围查询更快 2…...

Spring AI Chat Memory 实战指南:Local 与 JDBC 存储集成

一个面向 Java 开发者的 Sring-Ai 示例工程项目&#xff0c;该项目是一个 Spring AI 快速入门的样例工程项目&#xff0c;旨在通过一些小的案例展示 Spring AI 框架的核心功能和使用方法。 项目采用模块化设计&#xff0c;每个模块都专注于特定的功能领域&#xff0c;便于学习和…...