rabbitmq-spring-boot-start配置使用手册

rabbitmq-spring-boot-start配置使用手册

文章目录

  • 1.yaml配置如下
  • 2.引入pom依赖如下
    • 2.1 引入项目resources下libs中的jar包依赖如下
    • 2.2引入maven私服依赖如下
  • 3.启动类配置如下
  • 4.项目中测试发送消息如下
  • 5.项目中消费消息代码示例
  • 6.mq管理后台交换机队列创建及路由绑定关系如下

1.yaml配置如下

   rps中的每一个都可以按照Sping官方的RabbitAutoConfiguration自动装配的RabbitProperties的样式来配置,做到了实现配置多个rabbitMq服务器配置和一个rabbitMq服务器下可以配置多个不同类型的交换机和队列进行绑定,还实现了普通队列发送消息、

    一:延迟插件实现延迟队列

        交换机类型必须CustomExchange

    二:TTL + 死信队列/延迟交换机实现延迟队列

    三: 延迟交换机 + 消息设置setHeader(“x-delay”, xxx)

    以下配置了两个不同的rabbitMq服务器,每一个rabbitMq服务器对应多个队列,还配置了两个相同的rabbitMq服务器,只不过两个相同的rabbitMq服务器有不同的交换机,都是一个rabbitMq服务器可以对应相同的一套配置,代码功能服用性强,也方便快捷

## 配置需要保证唯一不重复(eqps中的每一的index唯一,一般配置成递增的,队列交换机绑定关系的bean注入都是根据rps的List下标+eqps中index下标注入保证了唯一性)
zlf:rabbit:rps:## 如果virtual-host不同,在配置一个即可,addresses不同也是可以在配置,eqps的下标以之对应上即可- rabbitmq:virtual-host: /dyict-uataddresses: 192.168.40.61port: 5672username: "admin"password: "admin"- rabbitmq:virtual-host: /testaddresses: 192.168.40.60port: 5672username: "admin"password: "admin"- rabbitmq:virtual-host: /test2addresses: 192.168.40.60port: 5672username: "admin"password: "admin"eqps:## 下标递增且唯一- index: 0eqs:- function-type: Delaydelay-type: 1exchange-type: customexchange-name: zlf.delay.test1queue-name: delay.test1routing-key: delay.test1.keyexchange-args:x-delayed-type: directqueue-args: {}- function-type: Normaldelay-type: 0exchange-type: directexchange-name: zlf.normal.test1queue-name: normal.test1routing-key: normal.test1.keyexchange-args: {}queue-args: {}- function-type: Delaydelay-type: 2exchange-type: directexchange-name: zlf.delay.test2queue-name: delay.test2## 不用监听正常的队列,直接根据同一个路由键去路由,然后监听死信队列routing-key: zlf.delay-test2-keydlx-exchange-name: zlf.dlx-test1dlx-exchange-type: directdlx-queue-name: dlx-test1dlx-key: zlf.dlx-test1-keyexchange-args: {}queue-args:x-dead-letter-exchange: zlf.dlx-test1x-dead-letter-routing-key: zlf.dlx-test1-key## 单位毫秒 30sx-message-ttl: 30000- function-type: Delaydelay-type: 3exchange-type: directexchange-name: zlf.delay.test3queue-name: delay.test3routing-key: zlf.delay-test3-keyexchange-args: {}queue-args: {}- index: 1eqs:- function-type: Delaydelay-type: 1exchange-type: customexchange-name: zlf.delay.test1queue-name: delay.test1routing-key: delay.test1.keyexchange-args:x-delayed-type: directqueue-args: {}- function-type: Normaldelay-type: 0exchange-type: directexchange-name: zlf.normal.test1queue-name: normal.test1routing-key: normal.test1.keyexchange-args: {}queue-args: {}- function-type: Delaydelay-type: 2exchange-type: directexchange-name: zlf.delay.test2queue-name: delay.test2## 不用监听正常的队列,直接根据同一个路由键去路由,然后监听死信队列routing-key: zlf.delay-test2-keydlx-exchange-name: zlf.dlx-test1dlx-exchange-type: directdlx-queue-name: dlx-test1dlx-key: zlf.dlx-test1-keyexchange-args: {}queue-args:x-dead-letter-exchange: zlf.dlx-test1x-dead-letter-routing-key: zlf.dlx-test1-key## 单位毫秒 30sx-message-ttl: 30000- function-type: Delaydelay-type: 3exchange-type: directexchange-name: zlf.delay.test3queue-name: delay.test3routing-key: zlf.delay-test3-keyexchange-args: {}queue-args: {}- index: 2eqs:- function-type: Delaydelay-type: 1exchange-type: customexchange-name: zlf.delay.test1queue-name: delay.test1routing-key: delay.test1.keyexchange-args:x-delayed-type: directqueue-args: {}- function-type: Normaldelay-type: 0exchange-type: directexchange-name: zlf.normal.test1queue-name: normal.test1routing-key: normal.test1.keyexchange-args: {}queue-args: {}- function-type: Delaydelay-type: 2exchange-type: directexchange-name: zlf.delay.test2queue-name: delay.test2## 不用监听正常的队列,直接根据同一个路由键去路由,然后监听死信队列routing-key: zlf.delay-test2-keydlx-exchange-name: zlf.dlx-test1dlx-exchange-type: directdlx-queue-name: dlx-test1dlx-key: zlf.dlx-test1-keyexchange-args: {}queue-args:x-dead-letter-exchange: zlf.dlx-test1x-dead-letter-routing-key: zlf.dlx-test1-key## 单位毫秒 30sx-message-ttl: 30000- function-type: Delaydelay-type: 3exchange-type: directexchange-name: zlf.delay.test3queue-name: delay.test3routing-key: zlf.delay-test3-keyexchange-args: {}queue-args: {}

2.引入pom依赖如下

2.1 引入项目resources下libs中的jar包依赖如下

    右键点击rabbitmq-spring-boot-start-1.0-SNAPSHOT.jar将该jar包手动导入(add as Library),复制该jar包到resources下libs,若果maven自动导入就不用右键手动导入

<dependency><groupId>org.zlf</groupId><artifactId>rabbitmq-spring-boot-start</artifactId><version>1.0-SNAPSHOT</version><scope>system</scope><systemPath>${pom.basedir}/src/main/resources/libs/rabbitmq-spring-boot-start-1.0-SNAPSHOT.jar</systemPath>
</dependency>

2.2引入maven私服依赖如下

<dependency><groupId>org.zlf</groupId><artifactId>rabbitmq-spring-boot-start</artifactId><version>1.0-SNAPSHOT</version>
</dependency>

3.启动类配置如下

@EnableZlfRabbitMq
@Import(value = {RabbitService.class, ZlfMqSpringUtils.class})
@SpringBootApplication(exclude = {RabbitAutoConfiguration.class})

4.项目中测试发送消息如下

    Controller测试可以根据rps的下标 + eqps的index下标来复制多个Controlle类,只需要调用api设置这两个下标对应解析即可发送

rabbitService.sendMsg6(0, 0, msg);

rabbitService.sendMsg6(1, 0, msg);

rabbitService.sendMsg6(2, 0, msg);

,,,,,,,,,,

上面配置了三个下标组合就有以下几种:

0 0 / 01 /11

1 0 / 11 /12

2 0 / 21 / 22

可以复制Controller1、Controller2、Controller3,,,,,,,调用时候只需要指定下标组合即可发送消息

package xxxx.controller;import com.dy.corporate.member.utils.SpringUtils;
import com.zlf.constants.ZlfMqRegistrarBeanNamePrefix;
import com.zlf.service.RabbitService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
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.RestController;@Slf4j
@RestController
@RequestMapping("rabbit")
public class RabbitMqTestController {@Autowiredprivate RabbitService rabbitService;@GetMapping("/sendDelayMsg")public String sendDelayMsg(@RequestParam(value = "msg") String msg) {log.info("sendDelayMsg.msg:{}", msg);RabbitTemplate rabbitTemplate = (RabbitTemplate) SpringUtils.getBean(ZlfMqRegistrarBeanNamePrefix.rabbitTemplatePrefix + 0);rabbitService.sendDelayed(rabbitTemplate, "zlf.delay.test1", "delay.test1.key", msg, 10);return "ok";}@GetMapping("/sendDelayMsg2")public String sendDelayMsg2(@RequestParam(value = "msg") String msg) {log.info("sendDelayMsg2.msg:{}", msg);rabbitService.sendMsg6(0, 0, msg);return "ok";}@GetMapping("/sendNormalMsg")public String sendNormalMsg(@RequestParam(value = "msg") String msg) {log.info("sendNormalMsg.msg:{}", msg);RabbitTemplate rabbitTemplate = (RabbitTemplate) SpringUtils.getBean(ZlfMqRegistrarBeanNamePrefix.rabbitTemplatePrefix + 0);rabbitService.sendMsg(rabbitTemplate, "zlf.delay.test1", "delay.test1.key", msg);return "ok";}@GetMapping("/sendNormalMsg2")public String sendNormalMsg2(@RequestParam(value = "msg") String msg) {log.info("sendNormalMsg2.msg:{}", msg);rabbitService.sendMsg6(0, 1, msg);return "ok";}@GetMapping("/sendDelayDlxMsg")public String sendDelayDlxMsg(@RequestParam(value = "msg") String msg) {log.info("sendDelayDlxMsg.msg:{}", msg);RabbitTemplate rabbitTemplate = (RabbitTemplate) SpringUtils.getBean(ZlfMqRegistrarBeanNamePrefix.rabbitTemplatePrefix + 0);//正常发延迟交换机和延迟队列的路由键rabbitService.sendDelayed(rabbitTemplate, "zlf.delay.test2", "zlf.delay-test2-key", msg, 10);return "ok";}@GetMapping("/sendDelayDlxMsg2")public String sendDelayDlxMsg2(@RequestParam(value = "msg") String msg) {log.info("sendDelayDlxMsg2.msg:{}", msg);//正常发延迟交换机和延迟队列的路由键rabbitService.sendDelayed6(0, 2, msg,10);return "ok";}@GetMapping("/sendDelayMsg3")public String sendDelayMsg3(@RequestParam(value = "msg") String msg) {log.info("sendDelayMsg3.msg:{}", msg);rabbitService.sendDelayed6(0, 3, msg, 10);return "ok";}}

项目utils下放入SpringUtils类:

package xxx.utils;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;/*** @author zlf* @description spring上下文工具类* @date 2024/03/11**/
@Component
public class SpringUtils implements ApplicationContextAware {private static final Logger logger = LoggerFactory.getLogger(SpringUtils.class);private static ApplicationContext applicationContext;@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {logger.info("应用程序上下文 : [{}]", "开始初始化");SpringUtils.applicationContext = applicationContext;logger.info("应用程序上下文 : [{}]", "初始化完成");}/*** 获取applicationContext*发给* @return*/public static ApplicationContext getApplicationContext() {return applicationContext;}/*** 通过name获取 Bean.** @param name* @return*/public static Object getBean(String name) {return getApplicationContext().getBean(name);}/*** 通过class获取Bean.** @param clazz* @param <T>* @return*/public static <T> T getBean(Class<T> clazz) {return getApplicationContext().getBean(clazz);}/*** 通过name,以及Clazz返回指定的Bean** @param name* @param clazz* @param <T>* @return*/public static <T> T getBean(String name, Class<T> clazz) {return getApplicationContext().getBean(name, clazz);}}

以上测试用例MqConsumer都是可以正常消费到对应队列中的消息的

5.项目中消费消息代码示例

    消费者中只需要指定对应的消费监听工厂即可,监听工厂配置如下:

ZlfMqRegistrarBeanNamePrefix.simpleRabbitListenerContainerFactory + rps的下标

可以复制多个MqConsumer1,MqConsumer2,,,,,,,,,

    然后指定对应的监听工厂配置下标即可,经过测试上面三个配置发送4中发送消息,监听消费都是正常的

package xxx.listener;import com.rabbitmq.client.Channel;
import com.zlf.constants.ZlfMqRegistrarBeanNamePrefix;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.io.IOException;/*** 手动ack业务demo* long deliveryTag = message.getMessageProperties().getDeliveryTag();* try {* int a = 1/0;* User user = JSONObject.parseObject(userStr,User.class);* log.info(user.toString());* //手动ack  第二个参数为false是表示仅仅确认当前消息 true表示确认之前所有的消息* channel.basicAck(deliveryTag,false);* } catch (Exception e) {* //手动nack 告诉rabbitmq该消息消费失败  第三个参数:如果被拒绝的消息应该被重新请求,而不是被丢弃或变成死信,则为true* try {* channel.basicNack(deliveryTag,false,false);* } catch (IOException ex) {* throw new RuntimeException("消息处理失败");* }* }* //channel.basicNack(); 不ack* //channel.basicReject(); 拒绝*/
@Slf4j
@Component
public class MqConsumer {/*** 延迟插件实现延迟队列监听队列消息** @param message* @param channel* @throws IOException*/@RabbitHandler@RabbitListener(queues = "delay.test1", containerFactory = ZlfMqRegistrarBeanNamePrefix.simpleRabbitListenerContainerFactory + 0)public void mqConsumer1(Message message, Channel channel) throws IOException {String msg = new String(message.getBody(), "UTF-8");try {log.info("mqConsumer1=====>msg:{}", msg);} catch (Exception e) {log.error("mqConsumer1消费异常:{}", e.getMessage());} finally {channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);}//channel.basicNack(); 不ack//channel.basicReject(); 拒绝}/*** 普通队列监听队列消息** @param message* @param channel* @throws IOException*/@RabbitHandler@RabbitListener(queues = "normal.test1", containerFactory = ZlfMqRegistrarBeanNamePrefix.simpleRabbitListenerContainerFactory + 0)public void mqConsumer2(Message message, Channel channel) throws IOException {String msg = new String(message.getBody(), "UTF-8");try {log.info("mqConsumer2=====>msg:{}", msg);} catch (Exception e) {log.error("mqConsumer2消费异常:{}", e.getMessage());} finally {channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);}}/*** TTL + 死信队列实现延迟队列监听延迟队列消息(此处省略)* 绑定的那个延迟队列,消息如果正常消费,则不会将消息投递到死信队列上,* 只有消息变成死信才会被投递到死信队列上*//*** TTL + 死信队列实现延迟队列监听死信队列消息* 成为死信的条件* * 1.队列消息长度到达限制。* * 2.消费者拒接消费消息,basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false。* * 3.原队列存在消息过期设置,消息到达超时时间未被消费。** @param message* @param channel* @throws IOException*/@RabbitHandler@RabbitListener(//监听连接工程指定containerFactory = ZlfMqRegistrarBeanNamePrefix.simpleRabbitListenerContainerFactory + 0,bindings = @QueueBinding(//延迟交换机exchange = @Exchange(value = "zlf.delay.test2",//持久化参数设置durable = "true",//交换机类型指定type = ExchangeTypes.DIRECT),//延迟交换机路由延迟队列的keykey = "zlf.delay-test2-key",//死信队列value = @Queue(value = "dlx-test1",//持久化参数设置durable = "true"//, //autoDelete = "false",)//ignoreDeclarationExceptions = "true")//,//concurrency = "1", // 指定监听该队列的消费者个数//ackMode = "MANUAL"// 手动ack)public void mqConsumer4(Message message, Channel channel) throws IOException {String msg = new String(message.getBody(), "UTF-8");try {log.info("mqConsumer4=====>msg:{}", msg);} catch (Exception e) {log.error("mqConsumer4:{}", e.getMessage());} finally {channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);}}/*** 延迟交换机 + 消息设置setHeader("x-delay", xxx)** @param message* @param channel* @throws IOException*/@RabbitHandler@RabbitListener(queues = "delay.test3", containerFactory = ZlfMqRegistrarBeanNamePrefix.simpleRabbitListenerContainerFactory + 0)public void mqConsumer5(Message message, Channel channel) throws IOException {String msg = new String(message.getBody(), "UTF-8");try {log.info("mqConsumer5=====>msg:{}", msg);} catch (Exception e) {log.error("mqConsumer5消费异常:{}", e.getMessage());} finally {channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);}//channel.basicNack(); 不ack//channel.basicReject(); 拒绝}}

6.mq管理后台交换机队列创建及路由绑定关系如下

    以下交换机和队列定义的时候都是持久化的,上面三个配置示都在rabbitMq的管理后台生成了相同的队列、交换机和绑定关系(唯一不一样的是错误对了是根据rps的List的下标来的,保证唯一),以下截图只展示一个即rps[0]的交换机、队列和绑定关系的创建,其它两个都是一样的(唯一不同就是错误队列交换机是后缀角标是递增的)。

zlf.delay.test1交换机(延迟插件)

img

zlf.delay.test1交换和delay.test1队列的绑定关系:

img

zlf.normal.test1交换机(普通交换机)

img

zlf.normal.test1交换机和normal.test1的绑定关系:

img

zlf.delay.test2交换机:(ttl + 死信队列)

img

zlf.delay.test2交换机和delay.test2绑定关系:

img

zlf.dlx-test1交换机:(死信交换机)

img

zlf.dlx-test1死信交换机和dlx-test1绑定关系:

img

zlf.delay.test3交换机:

img

zlf.delay.test3交换机delay.test3的绑定关系:

img

error交换机

error.direct + 一个下标,类型、路由键是固定

img

error.direct + 一个下标 和 error.queue+ 一个下标 的绑定关系:

img

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/536446.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Github 2024-03-13 开源项目日报 Top10

根据Github Trendings的统计,今日(2024-03-13统计)共有10个项目上榜。根据开发语言中项目的数量,汇总情况如下: 开发语言项目数量Rust项目2Python项目2非开发语言项目2HTML项目1CSS项目1Dart项目1TypeScript项目1Go项目1JavaScript项目1《Hello 算法》:动画图解、一键运行的…

【论文撰写】参考文献引用

1、交叉引用 当在撰写论文时往往需要在不同的位置引用参考文献&#xff0c;并且希望其在参考文献顺序与数量变动后也动态改变。这时&#xff0c;就需要word中的【引用】|【交叉引用】 在需要引用参考文献的地方点击后&#xff0c;选择对应的文献 将会得到以下效果 在将其设置为…

谷粒商城——分布式基础(全栈开发篇第一部分)

文章目录 一、服务治理网路数据支撑日志处理ELK应用监控集成工具开发工具 二、环境创建1、虚拟机创建2、虚拟机安装docker等1. 安装docker1. 配置阿里docker3.docker安装mysql错误 4、docker安装redis 3、软件1.Maven 阿里云镜像1.8jdk2、idea lombokmybatisX &#xff0c;3、 …

Python学习DAY13_图像处理

图形处理 用程序来处理图像和办公文档经常出现在实际开发中&#xff0c;Python 的标准库中虽然没有直接支持这些操作的模块&#xff0c;但我们可以通过 Python 生态圈中的第三方模块来完成这些操作。 相关知识 颜色 大家也许不太记得小时候美术课上的具体内容&#xff0c;但是…

【Flink SQL】Flink SQL 基础概念:SQL 动态表 连续查询

Flink SQL 基础概念&#xff1a;SQL 动态表 & 连续查询 1.SQL 应用于流处理的思路2.流批处理的异同点及将 SQL 应用于流处理核心解决的问题3.SQL 流处理的输入&#xff1a;输入流映射为 SQL 动态输入表4.SQL 流处理的计算&#xff1a;实时处理底层技术 - SQL 连续查询5.SQL…

css实现高度是宽度一半的效果

1、方法一&#xff1a;使用变量:root、var()、clac()实现&#xff1a; 1.1 效果如下&#xff1a; 2.2 代码如下&#xff1a; <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><title>Title</title>&l…

Python 常用的开源爬虫库介绍

Python 是一种广泛使用的编程语言&#xff0c;特别是在 Web 爬虫领域。有许多优秀的开源爬虫库可以帮助开发者高效地抓取网页内容。以下是几个常用的 Python 爬虫库及其特点和优势&#xff1a; BeautifulSoup 特点 - **HTML/XML 解析**&#xff1a;BeautifulSoup 是一个…

《如何使用C语言去下三子棋?》

目录 一、环境配置 二、功能模块 1.打印菜单 2.初始化并打印棋盘 3、行棋 3.1玩家行棋 3.2电脑行棋 4、判断是否和棋 5.判赢 三、代码实现 1、test.c文件 2、game.c文件 3、game.h文件 一、环境配置 本游戏用到三个文件&#xff0c;分别是两个源文件test.c game.c 和…

Python 数据持久层ORM框架 SQLAlchemy模块

文章目录 ORM 框架SQLAlchemy 简介SQLAlchemy 作用SQLAlchemy 原理SQLAlchemy 使用流程数据库驱动配置关系型数据库配置NoSQL数据库配置 创建引擎(Engine)定义模型类(ORM)创建会话(Session)创建数据库表其他关键字参数&#xff0c;这些参数将传递给底层的 DDL 创建函数。 删除数…

江科大stm32学习笔记——【6-1】TIM定时中断

一.定时器 1.类型 STM32F103C8T6定时器资源&#xff1a;TIM1,TIM2,TIM3,TIM4 我们这次用通用定时器 2.基本定时器 预分频器对来自RCC_TIMxCLK的72MHz进行预分频&#xff0c;比如写0就是不分频或1分频&#xff0c;输出频率输入频率72MHz。写1就是2分频&#xff0c;输出频率输…

基于Verilog的简易CPU设计

前言 本篇文章将简单讲解CPU之间各部分的功能及接线&#xff0c;并提供Verilog模拟CPU的各个组成部分。该CPU可以完成一些操作&#xff0c;如&#xff1a;加减法&#xff0c;与或&#xff0c;指令跳转等&#xff0c;最后提供testbench用于测试该CPU的工作情况是否符合预期。 C…

耳机壳UV树脂制作私模定制耳塞适合什么样的人使用呢?

耳机壳UV树脂制作私模定制耳塞适合什么样的人使用呢&#xff1f; 耳机壳UV树脂制作私模定制耳塞适合以下人群使用&#xff1a; 对音质要求高的人&#xff1a;私模定制耳塞能够完美契合用户的耳朵形状&#xff0c;减少漏音和外部噪音的干扰&#xff0c;提供更好的音质体验。需要…