SpringBoot学习（8）RabbitMQ详解

news2024/11/23 22:38:25

RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。

消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC 的调用等等。

以前一直使用的是 ActiveMQ，在实际的生产使用中也出现了一些小问题，在网络查阅了很多的资料后，决定尝试使用 RabbitMQ 来替换 ActiveMQ，RabbitMQ 的高可用性、高性能、灵活性等一些特点吸引了我们，查阅了一些资料整理出此文。

RabbitMQ 介绍

RabbitMQ 是实现 AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。 RabbitMQ 主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即 Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ 是一个开源的 AMQP 实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

相关概念

通常我们谈到队列服务, 会有三个概念： 发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和 队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。

                         

  • 左侧 P 代表 生产者，也就是往 RabbitMQ 发消息的程序。
  • 中间即是 RabbitMQ，其中包括了 交换机 和 队列。
  • 右侧 C 代表 消费者，也就是往 RabbitMQ 拿消息的程序。

那么，其中比较重要的概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。

  • 虚拟主机：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单， RabbitMQ 当中，用户只能在虚拟主机的粒度进行权限控制。 因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个 RabbitMQ 服务器都有一个默认的虚拟主机“/”。
  • 交换机：Exchange 用于转发消息，但是它不会做存储 ，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。
    这里有一个比较重要的概念：路由键 。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
  • 绑定：也就是交换机需要和队列相绑定，这其中如上图所示，是多对多的关系。

交换机(Exchange)

交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct, topic, Headers and Fanout

  • Direct：direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routing_key, 消息的routing_key 匹配时, 才会被交换器投送到绑定的队列中去.
  • Topic：按规则转发消息（最灵活）
  • Headers：设置 header attribute 参数类型的交换机
  • Fanout：转发消息到所有绑定队列

Direct Exchange

Direct Exchange 是 RabbitMQ 默认的交换机模式，也是最简单的模式，根据key全文匹配去寻找队列。

第一个 X - Q1 就有一个 binding key，名字为 orange； X - Q2 就有 2 个 binding key，名字为 black 和 green。当消息中的 路由键 和 这个 binding key 对应上的时候，那么就知道了该消息去到哪一个队列中。

Ps：为什么 X 到 Q2 要有 black，green，2个 binding key呢，一个不就行了吗？ - 这个主要是因为可能又有 Q3，而Q3只接受 black 的信息，而Q2不仅接受black 的信息，还接受 green 的信息。

Topic Exchange

Topic Exchange 转发消息主要是根据通配符。 在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

在这种交换机模式下：

  • 路由键必须是一串字符，用句号（.） 隔开，比如说 agreements.us，或者 agreements.eu.stockholm 等。
  • 路由模式必须包含一个 星号（*），主要用于匹配路由键指定位置的一个单词，比如说，一个路由模式是这样子：agreements..b.*，那么就只能匹配路由键是这样子的：第一个单词是 agreements，第四个单词是 b。 井号（#）就表示相当于一个或者多个单词，例如一个匹配模式是 agreements.eu.berlin.#，那么，以agreements.eu.berlin 开头的路由键都是可以的。

具体代码发送的时候还是一样，第一个参数表示交换机，第二个参数表示 routing key，第三个参数即消息。如下：

rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2", " this is  RabbitMQ!");

topic 和 direct 类似, 只是匹配上支持了"模式", 在"点分"的 routing_key 形式中, 可以使用两个通配符:

  • *表示一个词.
  • #表示零个或多个词.

Headers Exchange

headers 也是根据规则匹配, 相较于 direct 和 topic 固定地使用 routing_key , headers 则是一个自定义匹配规则的类型.
在队列与交换器绑定时, 会设定一组键值对规则, 消息中也包括一组键值对( headers 属性), 当这些键值对有一对, 或全部匹配时, 消息被投送到对应队列.

Fanout Exchange

Fanout Exchange 消息广播的模式，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了 routing_key 会被忽略。

Spring Boot 集成 RabbitMQ

Spring Boot 集成 RabbitMQ 非常简单，如果只是简单的使用配置非常少，Spring Boot 提供了spring-boot-starter-amqp 项目对消息各种支持。

简单使用

1、配置 Pom 包，主要是添加 spring-boot-starter-amqp 的支持

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、配置文件

配置 RabbitMQ 的安装地址、端口以及账户信息

spring.application.name=Spring-boot-rabbitmq

spring.rabbitmq.host=192.168.0.86
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

3、队列配置

@Configuration
public class RabbitConfig {

    @Bean
    public Queue Queue() {
        return new Queue("hello");
    }

}

3、发送者

rabbitTemplate 是 Spring Boot 提供的默认实现

@component
public class HelloSender {

	@Autowired
	private AmqpTemplate rabbitTemplate;

	public void send() {
		String context = "hello " + new Date();
		System.out.println("Sender : " + context);
		this.rabbitTemplate.convertAndSend("hello", context);
	}

}

4、接收者

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {

    @RabbitHandler
    public void process(String hello) {
        System.out.println("Receiver  : " + hello);
    }

}

5、测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest {

	@Autowired
	private HelloSender helloSender;

	@Test
	public void hello() throws Exception {
		helloSender.send();
	}

}

注意，发送者和接收者的 queue name 必须一致，不然不能接收

多对多使用

一个发送者，N 个接收者或者 N 个发送者和 N 个接收者会出现什么情况呢？

一对多发送

对上面的代码进行了小改造，接收端注册了两个 Receiver,Receiver1 和 Receiver2，发送端加入参数计数，接收端打印接收到的参数，下面是测试代码，发送一百条消息，来观察两个接收端的执行效果

@Test
public void oneToMany() throws Exception {
	for (int i=0;i<100;i++){
		neoSender.send(i);
	}
}

结果如下：

Receiver 1: Spring boot neo queue ****** 11
Receiver 2: Spring boot neo queue ****** 12
Receiver 2: Spring boot neo queue ****** 14
Receiver 1: Spring boot neo queue ****** 13
Receiver 2: Spring boot neo queue ****** 15
Receiver 1: Spring boot neo queue ****** 16
Receiver 1: Spring boot neo queue ****** 18
Receiver 2: Spring boot neo queue ****** 17
Receiver 2: Spring boot neo queue ****** 19
Receiver 1: Spring boot neo queue ****** 20

根据返回结果得到以下结论

一个发送者，N个接受者,经过测试会均匀的将消息发送到N个接收者中

多对多发送

复制了一份发送者，加入标记，在一百个循环中相互交替发送

@Test
	public void manyToMany() throws Exception {
		for (int i=0;i<100;i++){
			neoSender.send(i);
			neoSender2.send(i);
		}
}

结果如下：

Receiver 1: Spring boot neo queue ****** 20
Receiver 2: Spring boot neo queue ****** 20
Receiver 1: Spring boot neo queue ****** 21
Receiver 2: Spring boot neo queue ****** 21
Receiver 1: Spring boot neo queue ****** 22
Receiver 2: Spring boot neo queue ****** 22
Receiver 1: Spring boot neo queue ****** 23
Receiver 2: Spring boot neo queue ****** 23
Receiver 1: Spring boot neo queue ****** 24
Receiver 2: Spring boot neo queue ****** 24
Receiver 1: Spring boot neo queue ****** 25
Receiver 2: Spring boot neo queue ****** 25

结论：和一对多一样，接收端仍然会均匀接收到消息

高级使用

对象的支持

Spring Boot 以及完美的支持对象的发送和接收，不需要格外的配置。

//发送者
public void send(User user) {
	System.out.println("Sender object: " + user.toString());
	this.rabbitTemplate.convertAndSend("object", user);
}

...

//接收者
@RabbitHandler
public void process(User user) {
    System.out.println("Receiver object : " + user);
}

结果如下：

Sender object: User{name='neo', pass='123456'}
Receiver object : User{name='neo', pass='123456'}

Topic Exchange

topic 是 RabbitMQ 中最灵活的一种方式，可以根据 routing_key 自由的绑定不同的队列

首先对 topic 规则配置，这里使用两个队列来测试

@Configuration
public class TopicRabbitConfig {

    final static String message = "topic.message";
    final static String messages = "topic.messages";

    @Bean
    public Queue queueMessage() {
        return new Queue(TopicRabbitConfig.message);
    }

    @Bean
    public Queue queueMessages() {
        return new Queue(TopicRabbitConfig.messages);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("exchange");
    }

    @Bean
    Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
    }

    @Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }
}

使用 queueMessages 同时匹配两个队列，queueMessage 只匹配 "topic.message" 队列

public void send1() {
	String context = "hi, i am message 1";
	System.out.println("Sender : " + context);
	this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
}

public void send2() {
	String context = "hi, i am messages 2";
	System.out.println("Sender : " + context);
	this.rabbitTemplate.convertAndSend("exchange", "topic.messages", context);
}

发送send1会匹配到topic.#和topic.message 两个Receiver都可以收到消息，发送send2只有topic.#可以匹配所有只有Receiver2监听到消息

Fanout Exchange

Fanout 就是我们熟悉的广播模式或者订阅模式，给 Fanout 交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

Fanout 相关配置

@Configuration
public class FanoutRabbitConfig {

    @Bean
    public Queue AMessage() {
        return new Queue("fanout.A");
    }

    @Bean
    public Queue BMessage() {
        return new Queue("fanout.B");
    }

    @Bean
    public Queue CMessage() {
        return new Queue("fanout.C");
    }

    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    @Bean
    Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(AMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(BMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(CMessage).to(fanoutExchange);
    }

}

这里使用了 A、B、C 三个队列绑定到 Fanout 交换机上面，发送端的 routing_key 写任何字符都会被忽略：

public void send() {
	String context = "hi, fanout msg ";
	System.out.println("Sender : " + context);
	this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}

结果如下：

Sender : hi, fanout msg 
...
fanout Receiver B: hi, fanout msg 
fanout Receiver A  : hi, fanout msg 
fanout Receiver C: hi, fanout msg

结果说明，绑定到 fanout 交换机上面的队列都收到了消息

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

搞定JavaScript异步原理，深入学习Promise

搞定JavaScript异步原理，深入学习Promise

JS异步及Promise基础 一、从浏览器运行机制说起 来破案啦&#xff0c;常说的“多线程的浏览器、单线程的Javascript”到底是什么意思&#xff1f;单线程的Javascript是否真的需要抽身去计时、去发http请求&#xff1f;第一部分我们重新从浏览器运行机制开始认识一下JavaScript…
阅读更多...
828华为云征文 | Flexus X实例与华为云EulerOS的Tomcat安装指南

828华为云征文 | Flexus X实例与华为云EulerOS的Tomcat安装指南

文章目录 前言安全组设置操作步骤软件安装配置软件验证Tomcat安装是否成功 总结 前言 Tomcat是一个由Apache软件基金会开发并维护的免费、开源的Web应用服务器。它主要用于处理Java Servlet、JavaServer Pages&#xff08;JSP&#xff09;和JavaServer Pages Standard Tag Lib…
阅读更多...
新增一个数组传递给后端

新增一个数组传递给后端

实现的效果&#xff1a; 页面 <div style"margin-bottom: 10px" v-if"totalPrice"><p style"font-weight: bolder;margin-bottom: 10px">支付计划<el-button type"text" size"small" click"addPayInf…
阅读更多...
53 mysql pid 文件的创建

53 mysql pid 文件的创建

前言 接上一篇文章 mysql 启动过程中常见的相关报错信息 在 mysql 中文我们在 “service mysql start”, “service mysql stop” 经常会碰到 mysql.pid 相关的错误信息 比如 “The server quit without updating PID file” 我们这里来看一下 mysql 中 mysql.pid 文件的…
阅读更多...
软件工程知识点总结（2）：需求分析（一）——用例建模

软件工程知识点总结（2）：需求分析（一）——用例建模

1 软件项目开发流程&#xff1a; 需求分析→概要设计→详细设计→编码实施→测试→产品提 交→维护 2 系统必须做什么&#xff1f; 获取用户需求&#xff0c;从用户角度考虑&#xff0c;用户需要系统必须完成哪些工作&#xff0c;也就是对目 标系统提出完整、准确、清晰、具体…
阅读更多...
算法day22|组合总和 （含剪枝）、40.组合总和II、131.分割回文串

算法day22|组合总和 （含剪枝）、40.组合总和II、131.分割回文串

算法day22|组合总和 &#xff08;含剪枝&#xff09;、40.组合总和II、131.分割回文串 39. 组合总和 &#xff08;含剪枝&#xff09;40.组合总和II131.分割回文串 39. 组合总和 &#xff08;含剪枝&#xff09; 给你一个 无重复元素 的整数数组 candidates 和一个目标整数 ta…
阅读更多...
使用Node-API实现跨语言交互开发流程

使用Node-API实现跨语言交互开发流程

一、前言 使用Node-API实现跨语言交互&#xff0c;首先需要按照Node-API的机制实现模块的注册和加载等相关动作。 ArkTS/JS侧&#xff1a;实现C方法的调用。代码比较简单&#xff0c;import一个对应的so库后&#xff0c;即可调用C方法。 Native侧&#xff1a;.cpp文件&#xf…
阅读更多...
【C语言】十六进制、二进制、字节、位、指针、数组

【C语言】十六进制、二进制、字节、位、指针、数组

【C语言】十六进制、二进制、字节、位 文章目录 [TOC](文章目录) 前言一、十六进制、二进制、字节、位二、变量、指针、指针变量三、数组与指针四、指针自加运算五、二维数组与指针六、指向指针的指针七、指针变量作为函数形参八、函数指针九、函数指针数组十、参考文献总结 前…
阅读更多...
JVM面试（六）垃圾收集器

JVM面试（六）垃圾收集器

目录 概述STW收集器的并发和并行 Serial收集器ParNew收集器Parallel Scavenge收集器Serial Old收集器Parallel Old收集器CMS收集器Garbage First&#xff08;G1&#xff09;收集器 概述 上一章我们分析了垃圾收集算法&#xff0c;那这一章我们来认识一下这些垃圾收集器是如何运…
阅读更多...
某云彩SRM2.0任意文件下载漏洞

某云彩SRM2.0任意文件下载漏洞

文章目录 免责申明搜索语法漏洞描述漏洞复现修复建议 免责申明 本文章仅供学习与交流&#xff0c;请勿用于非法用途&#xff0c;均由使用者本人负责&#xff0c;文章作者不为此承担任何责任 搜索语法 fofa icon_hash"1665918155"漏洞描述 某云采 SRM2.0是一款先…
阅读更多...
制作自己的游戏：打砖块

制作自己的游戏：打砖块

文章目录 &#x1f680; 前言&#x1f680; 前期准备&#x1f680; 玩法设计&#x1f680; 游戏场景&#x1f353; 什么是游戏场景&#x1f353; 绘制左上角积分&#x1f353; 绘制右上角生命值&#x1f353; 绘制砖块&#x1f353; 绘制小球&#x1f353; 绘制挡板&#x1f35…
阅读更多...
场景是人工智能第四要素，是垂直领域人工智能的第一要素。

场景是人工智能第四要素，是垂直领域人工智能的第一要素。

"场景是人工智能的第四要素&#xff0c;与数据、算力、算法同等重要。"拿着技术找场景&#xff0c;还是拿着场景找技术&#xff1f;这个锤子和钉子的问题&#xff0c;一直困扰着各家AI大厂。从近5年的实践来看&#xff0c;拿着场景找技术是更为稳健的&#xff0c;否则…
阅读更多...
文章解读与仿真程序复现思路——电力自动化设备EI\CSCD\北大核心《面向电网调峰的聚合温控负荷多目标优化控制方法 》

文章解读与仿真程序复现思路——电力自动化设备EI\CSCD\北大核心《面向电网调峰的聚合温控负荷多目标优化控制方法 》

本专栏栏目提供文章与程序复现思路&#xff0c;具体已有的论文与论文源程序可翻阅本博主免费的专栏栏目《论文与完整程序》 论文与完整源程序_电网论文源程序的博客-CSDN博客https://blog.csdn.net/liang674027206/category_12531414.html 电网论文源程序-CSDN博客电网论文源…
阅读更多...
R901085689比例流量控制阀配置HE-SP1比例放大器

R901085689比例流量控制阀配置HE-SP1比例放大器

R901085689比例流量控制阀配置HE-SP1比例放大器的功能是将电信号转换成对应的流量变化&#xff0c;通过调整阀门开度来控制介质的流量。这种转换是通过比例电磁铁实现的&#xff0c;它将输入的电流信号转换成力或位移&#xff0c;从而驱动阀芯移动&#xff0c;实现流量的连续调…
阅读更多...
html+css+js网页设计 珠宝首饰模版13个页面

html+css+js网页设计 珠宝首饰模版13个页面

htmlcssjs网页设计 珠宝首饰模版13个页面 网页作品代码简单&#xff0c;可使用任意HTML编辑软件&#xff08;如&#xff1a;Dreamweaver、HBuilder、Vscode 、Sublime 、Webstorm、Text 、Notepad 等任意html编辑软件进行运行及修改编辑等操作&#xff09;。 获取源码 1&…
阅读更多...
手机同时传输USB功能与充电的实现及LDR6500的作用

手机同时传输USB功能与充电的实现及LDR6500的作用

在智能设备日益普及的今天&#xff0c;用户对于手机的功能需求愈发多样化&#xff0c;其中同时实现USB数据传输与充电功能成为了许多用户的迫切需求。这一功能的实现离不开先进的硬件技术和创新的芯片解决方案&#xff0c;而LDR6500正是这样一款能够满足这一需求的USB PD&#…
阅读更多...
uni-app 扫码优化：谈谈我是如何提升安卓 App 扫码准确率的

uni-app 扫码优化：谈谈我是如何提升安卓 App 扫码准确率的

一. 前言 之前的一个项目遭到用户吐槽&#xff1a;“你们这个 App 扫码的正确率太低了&#xff0c;尤其是安卓的设备。经常性的扫码扫不出来&#xff0c;就算是扫出来了&#xff0c;也是错误的结果&#xff01;” 由于之前是扫描二维码的需求&#xff0c;所以没有对扫描条形码…
阅读更多...
yolov8-obb旋转目标检测onnxruntime和tensorrt推理

yolov8-obb旋转目标检测onnxruntime和tensorrt推理

onnxruntime推理 导出onnx模型&#xff1a; from ultralytics import YOLO model YOLO("yolov8n-obb.pt") model.export(format"onnx") onnx模型结构如下&#xff1a; python推理代码&#xff1a; import cv2 import math import numpy as np impo…
阅读更多...
全面提升管理效率的智慧园区可视化系统

全面提升管理效率的智慧园区可视化系统

通过图扑 HT 搭建智慧园区可视化&#xff0c;实时监测和展示园区内各设施的状态与能耗&#xff0c;优化资源配置&#xff0c;提升园区管理效率。
阅读更多...
科普神文，一次性讲透AI大模型的核心概念

科普神文，一次性讲透AI大模型的核心概念

令牌&#xff0c;向量&#xff0c;嵌入&#xff0c;注意力&#xff0c;这些AI大模型名词是否一直让你感觉熟悉又陌生&#xff0c;如果答案肯定的话&#xff0c;那么朋友&#xff0c;今天这篇科普神文不容错过。我将结合大量示例及可视化的图形手段&#xff0c;为你由浅入深一次…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023