SpringBoot 整合 RabbitMQ 实现流量消峰

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

消息中间件在互联网公司的使用中越来越多,刚才还看到新闻阿里将 RocketMQ 捐献给了 Apache,当然了今天的主角还是讲 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 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routingkey, 消息的routingkey 匹配时, 才会被交换器投送到绑定的队列中去.

  • 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-rabbitmqspring.rabbitmq.host=192.168.0.86
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

3、队列配置

@Configuration
public class RabbitConfig {@Beanpublic Queue Queue() {return new Queue("hello");}}

4、发送者

rabbitTemplate 是 Spring Boot 提供的默认实现

@component
public class HelloSender {@Autowiredprivate 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 {@RabbitHandlerpublic void process(String hello) {System.out.println("Receiver  : " + hello);}}

5、测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest {@Autowiredprivate HelloSender helloSender;@Testpublic 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 test queue ****** 11
Receiver 2: Spring boot test queue ****** 12
Receiver 2: Spring boot test queue ****** 14
Receiver 1: Spring boot test queue ****** 13
Receiver 2: Spring boot test queue ****** 15
Receiver 1: Spring boot test queue ****** 16
Receiver 1: Spring boot test queue ****** 18
Receiver 2: Spring boot test queue ****** 17
Receiver 2: Spring boot test queue ****** 19
Receiver 1: Spring boot test 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 test queue ****** 20
Receiver 2: Spring boot test queue ****** 20
Receiver 1: Spring boot test queue ****** 21
Receiver 2: Spring boot test queue ****** 21
Receiver 1: Spring boot test queue ****** 22
Receiver 2: Spring boot test queue ****** 22
Receiver 1: Spring boot test queue ****** 23
Receiver 2: Spring boot test queue ****** 23
Receiver 1: Spring boot test queue ****** 24
Receiver 2: Spring boot test queue ****** 24
Receiver 1: Spring boot test queue ****** 25
Receiver 2: Spring boot test 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='test', pass='123456'}
Receiver object : User{name='test', 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";@Beanpublic Queue queueMessage() {return new Queue(TopicRabbitConfig.message);}@Beanpublic Queue queueMessages() {return new Queue(TopicRabbitConfig.messages);}@BeanTopicExchange exchange() {return new TopicExchange("exchange");}@BeanBinding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");}@BeanBinding 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 {@Beanpublic Queue AMessage() {return new Queue("fanout.A");}@Beanpublic Queue BMessage() {return new Queue("fanout.B");}@Beanpublic Queue CMessage() {return new Queue("fanout.C");}@BeanFanoutExchange fanoutExchange() {return new FanoutExchange("fanoutExchange");}@BeanBinding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {return BindingBuilder.bind(AMessage).to(fanoutExchange);}@BeanBinding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {return BindingBuilder.bind(BMessage).to(fanoutExchange);}@BeanBinding 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.mzph.cn/pingmian/64013.shtml

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

相关文章

Git-分支(branch)常用命令

分支 我们在做项目开发的时候&#xff0c;无论是软件项目还是其他机械工程项目&#xff0c;我们为了提高效率以及合理的节省时间等等原因&#xff0c;现在都不再是线性进行&#xff0c;而是将一个项目抽离出诸进行线&#xff0c;每一条线在git中我们就叫做分支&#xff0c;bran…

springboot425滑雪场管理系统(论文+源码)_kaic

摘要 近年来&#xff0c;信息化管理行业的不断兴起&#xff0c;使得人们的日常生活越来越离不开计算机和互联网技术。首先&#xff0c;根据收集到的用户需求分析&#xff0c;对设计系统有一个初步的认识与了解&#xff0c;确定滑雪场管理系统的总体功能模块。然后&#xff0c;详…

好玩的汇编编译器NASM:一款基于x86架构的汇编与反汇编软件

好玩的汇编编译器NASM This is the project webpage for the Netwide Assembler (NASM), an asssembler for the x86 CPU architecture portable to nearly every modern platform, and with code generation for many platforms old and new. Netwide Assembler&#xff08;…

【AI知识】有监督学习分类任务之支持向量机

1.支持向量机概念 支持向量机&#xff08;Support Vector Machine, SVM&#xff09; 是一种有监督学习算法&#xff0c;主要用于分类任务&#xff08;也可用于回归任务&#xff0c;即支持向量回归&#xff0c;SVR&#xff09;。SVM的核心思想是找到一个最优的超平面&#xff0…

杭州乘云联合信通院发布《云计算智能化可观测性能力成熟度模型》

原文地址&#xff1a;杭州乘云联合中国信通院等单位正式发布《云计算智能化可观测性能力成熟度模型》标准 2024年12月3日&#xff0c;由全球数字经济大会组委会主办、中国信通院承办的 2024全球数字经济大会 云AI计算创新发展大会&#xff08;2024 Cloud AI Compute Ignite&…

【经典】制造供应链四类策略(MTS、MTO、ATO、ETO)细说

关注作者 制造供应链的牛鞭问题与复杂问题主要是从两个方面解决&#xff0c;一是同步化供应链消减从需求到供应的放大效应&#xff0c;二是供应链细分&#xff0c;针对不同的客户、不同的需求供应的匹配策略来应对复杂性&#xff0c;更好的满足客户并以最低的总成本来实现。 对…

实时日志与发展:Elasticsearch 推出全新专用的 logsdb 索引模式

作者&#xff1a;来自 Elastic Mark Settle, George Kobar 及 Amena Siddiqi Elastic 最新发布的 logsdb 索引模式是专为日志管理优化的功能&#xff0c;旨在提升日志数据的存储效率、查询性能以及整体可用性。这个模式专注于满足现代日志处理需求&#xff0c;提供更高效的日志…

React Image Crop——在React应用中轻松实现图片裁剪功能

React Image Crop是一个用于在React应用程序中裁剪和调整图像的库。它提供了一个简单而强大的界面&#xff0c;允许用户选择和调整裁剪区域&#xff0c;并生成裁剪后的图像。 什么是React Image Crop&#xff1f; React Image Crop是一个开源的React组件&#xff0c;用于在浏览…

【HarmonyOS】鸿蒙应用实现手机摇一摇功能

【HarmonyOS】鸿蒙应用实现手机摇一摇功能 一、前言 手机摇一摇功能&#xff0c;是通过获取手机设备&#xff0c;加速度传感器接口&#xff0c;获取其中的数值&#xff0c;进行逻辑判断实现的功能。 在鸿蒙中手机设备传感器ohos.sensor (传感器)的系统API监听有以下&#xf…

微积分复习笔记 Calculus Volume 2 - 3.7 Improper Integrals

3.7 Improper Integrals - Calculus Volume 2 | OpenStax

ViewModel

ViewMode是MVVM架构模式中VM层对应的类&#xff0c;它的作用是存储界面数据&#xff0c;并和界面发生数据交互。ViewModel能感知生命周期&#xff0c;并且在界面由于配置问题发生重建时候&#xff0c;可以保持当前的数据不变。生命周期如下&#xff1a; ViewMode由ViewModePr…

智慧城市道路分割数据集labelme格式2648张1类别

数据集格式&#xff1a;labelme格式(不包含mask文件&#xff0c;仅仅包含jpg图片和对应的json文件) 图片数量(jpg文件个数)&#xff1a;2648 标注数量(json文件个数)&#xff1a;2648 标注类别数&#xff1a;1 标注类别名称:["road"] 每个类别标注的框数&#x…

leetcode-73.矩阵置零-day5

class Solution {public void setZeroes(int[][] mat) {int m mat.length, n mat[0].length;// 1. 扫描「首行」和「首列」记录「首行」和「首列」是否该被置零boolean r0 false, c0 false;for (int i 0; i < m; i) {if (mat[i][0] 0) {r0 true;break;}}for (int j …

uni-app H5端使用注意事项 【跨端开发系列】

&#x1f517; uniapp 跨端开发系列文章&#xff1a;&#x1f380;&#x1f380;&#x1f380; uni-app 组成和跨端原理 【跨端开发系列】 uni-app 各端差异注意事项 【跨端开发系列】uni-app 离线本地存储方案 【跨端开发系列】uni-app UI库、框架、组件选型指南 【跨端开…

DCDC降压模块

一、项目介绍 在B站看到工科男孙老师发的DCDC模块视频&#xff0c;后面有提到这款芯片&#xff0c;于是就使用这款芯片自己设计了这款降压模块。 芯片的封装为TSOT23-8&#xff0c;对空间利用率高。 输入电压的范围为3-18V。 输出电压的范围为0-8V。 输出电流最…

康耐视智能相机(Insight)通过ModbusTCP发送字符串到倍福(BECKHOFF)PLC中

文章目录 1.背景2.分析3.实现3.1.PLC的ModbusTCP_Server3.1.1.安装TF6250-Modbus-TCP3.1.2.PLC设置 3.2.智能相机的ModbusTCP_Client3.2.1.了解ModbusTCP的协议3.2.2.根据协议写代码3.2.2.1.纯函数代码3.2.2.2.脚本代码 3.2.3.非脚本处理时的代码逻辑图3.2.4.关于代码的问题及解…

基于多视角深度学习技术的乳腺X线分类:图神经网络与Transformer架构的研究|文献速递-生成式模型与transformer在医学影像中的应用速递

Title 题目 Mammography classification with multi-view deep learning techniques:Investigating graph and transformer-based architectures 基于多视角深度学习技术的乳腺X线分类&#xff1a;图神经网络与Transformer架构的研究 01 文献速递介绍 乳腺X线检查是乳腺癌…

恢复删除的文件:6个免费Windows电脑数据恢复软件

数据恢复软件可帮助您从众多存储设备中恢复损坏或删除的数据。您可以使用这些文件恢复软件来检索文件、文档、视频、图片等。这些应用程序支持多种标准文件格式&#xff0c;如 PNG、RTF、PDF、HTML、JPG、MP3 等。 经过超过 75 小时的研究&#xff0c;我分析了 25 最佳免费数据…

探索 Python 编程调试案例:程序平均值的修复过程

&#x1f49d;&#x1f49d;&#x1f49d;Python 作为一门广泛应用的编程语言&#xff0c;其编程过程并非总是一帆风顺。即使是经验丰富的程序员&#xff0c;也会在代码中遇到各种错误。而调试Python代码过程&#xff0c;也是学习中不可避免的步骤。 今天来看一个简单的案例。…

简单了解一下 Go 语言构建约束?

​构建约束是一种在 Go 语言中控制源文件编译条件的方法&#xff0c;它可以让您指定某些文件只在特定的操作系统、架构、编译器或 Go 版本下编译&#xff0c;而在其他环境中自动忽略。这样可以方便您针对不同的平台或场景编写不同的代码&#xff0c;实现条件编译的功能。 构建…