SpringBoot RabbitMQ收发消息、配置及原理

今天分析SpringBoot通过自动配置集成RabbitMQ的原理以及使用。

AMQP概念

RabbitMQ是基于AMQP协议的message broker，所以我们首先要对AMQP做一个简单的了解。

AMQP (Advanced Message Queuing Protocol) is a messaging protocol that enables conforming client applications to communicate with conforming messaging middleware brokers.

AMQP是Advanced Message Queuing Protocol的缩写，是一个消息收发客户端与消息中间件都要遵守的消息通讯协议。

AMQP消息模型如下图：
在这里插入图片描述
消息有发布者发送给Exchange，Exchange通过绑定路由规则路由该消息到消息队列中，之后消息中间件投递该消息给指定的消费者、或者由消费者按需拉取/消费消息。

消息丢失问题：网络原因、或者消费者消费逻辑的原因均可能导致消息发送（接收）失败，AMQP通过消息确认（message acknowledgements）机制来处理消息发送失败：消息送达消费者后，消费者需发送确认消息通知broker。启用消息确认机制后，broker必须在收到确认之后才能从队列中移除消息，确保消息不丢失。

消息无法送达问题：消息无法送达消费者的情况下，该消息会根据参数设置被broker返回给消息发送者、直接丢弃、或者丢入“死信”队列。

Exchange Type

AMQP消息发送给Exchange，Exchange接收到消息后路由给0个或多个队列，路由规则依赖于Exchange Type。

AMQP包含如下Exange Type：

  1. Direct exchange：路由匹配后直接投递
  2. Fanout exchange：消息投递给绑定到当前Exchange的所有队列（不需要匹配）。
  3. Topic exchange：根据订阅主题投递消息。
  4. Headers exchange：路由规则定义在消息头信息中，根据头信息匹配投递。

队列

队列是AMQP中存储消息的地方，队列具有如下属性：

  1. Name：队列名称。每一个队列都有名称，不指定名称则broker会自动生成一个名称。"amq."开头的队列名称是AMQP的保留名称，不允许应用层使用。
  2. Durable (the queue will survive a broker restart)：存活周期。队列创建的时候可以被声明为持久队列或临时队列，持久队列的元数据存储在硬盘、而临时队列存储在内存。
  3. Exclusive (used by only one connection and the queue will be deleted when that connection closes)：独占队列，连接关闭后队列删除。
  4. Auto-delete (queue that has had at least one consumer is deleted when last consumer unsubscribes)：队列必须绑定至少一个消费者，没有消费者绑定到给队列则队列自动删除。
  5. Arguments (optional; used by plugins and broker-specific features such as message TTL, queue length limit, etc)：其他消息参数。

队列使用之前必须首先声明（通过声明指定消息队列参数），声明队列等同于创建队列，声明队列时如果队列已经存在，则：声明的参数与已存在队列参数相同，对当前已存在队列没有任何影响，如果声明的参数与已存在队列参数不同，抛出异常。

绑定

绑定指的是Exchange路由消息到队列的规则。队列首先必须绑定到Exchange之后，Exchange才能发送消息到该队列。routing key是绑定的可选属性，特定Exchange（Direct exchange）发送消息到队列时会根据routing key进行匹配。

消费者

消费者通过以下两种方式从队列中消费消息：

  1. Push方式：消息主动投递给消费者（推荐方式）。
  2. Poll方式：消费者轮询获取（性能原因，不推荐）。

（(Consumer) Delivery Acknowledgements）投递确认

由于网络原因，或者消费者应用在消息消费过程中发生错误导致消息消费失败------均可导致消息丢失。

为了尽可能避免消息丢失，AMQP提供了两种消息确认机制：

  1. 自动确认模式：broker发送消息之后确认。
  2. 显式确认：消费者反馈确认消息。

显式确认模式下，由消费者决定确认消息的反馈时机：收到消息后立即反馈、或者消息持久化之后反馈、或者消息完全被消费（消费业务处理完成）之后反馈。

如果消费者一直不给broker发送确认信息，broker会重新投递消息给其他消费者、如果当前没有其他消费者注册到当前队列的话，broker会一直保存该消息直到有新的消费者注册上来之后发送。

拒信

消费者可给broker返回拒绝消息指令，以告知broker消费者处理消息失败。根据消费者的指令，broker可以丢弃该消息、或者重新进入队列等待再次消费。进入队列等待再次消费的情况下，消费者需注意避免造成该消息的循环投递。

Prefetching Messages

多个消费者共享同一队列的场景下，broker可以控制每一消费者最多可以并发处理的消息数量（消费者返回消息确认前brokenr最多投递到该消费者的消息数量）。这种机制其实是一种简单的load balance，一个消费者如果长时间不返回消息确认，极有可能说明该消费者的消费能力出现问题。

Connections

与数据库连接类似，AMQP客户端（生产者或者消费者）通过connections连接broker、并通过connection收发消息。

Channels

多数场景下，客户端都需要创建多个连接以便并发处理消息，但是同时创建多个TCP连接在性能和网络管理上都很不划算，因此，AMQP通过channels多路复用TCP连接：一个连接可以包含多个Channels，所有的客户端操作都依赖于Channels执行，每一个Channels都独立运行互不影响。

RabbitMQ

RabbitMQ is the most widely deployed open source message broker.

RabbitMQ是基于AMQP协议的、Erlang开发的被广泛部署的开源消息队列。

有关RabbitMQ的详细特性本文不展开讨论。

RabbitMQ的安装

RabbitMQ基于Erlang，所以安装RabbitMQ之前，你必须已经安装好了Erlang。不同版本的RabbitMQ对Erlang的版本也有不同的要求，详情参见https://www.rabbitmq.com/which-erlang.html。

根据下载安装Erlang/OTP 25.3.2。

之后从RabbitMQ官网（https://www.rabbitmq.com/install-windows.html）下载安装RabbitMQ（rabbitmq-server-3.12.12.exe）。

SpringBoot项目配置RabbitMQ

RabbitMQ is a lightweight, reliable, scalable, and portable message broker based on the AMQP protocol. Spring uses RabbitMQ to communicate through the AMQP protocol.

Spring使用RabbitMQ作为AMQP协议下的消息队列broker，SpringBoot提供了对RabbitMQ的基于autoConfiguration的支持。

pom文件引入RabbitMQ

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

yml文件配置

application.yml文件中配置RabbitMQ：

spring:rabbitmq:host: localhostport: 5672username: guestpassword: guestvirtual-host: /#确认消息已发送到交换机（Exchange）publisher-confirm-type: correlated#确认消息已发送到队列（Queue）publisher-returns: true#消费方消息确认listener:simple:#确认方式：手动确认acknowledge-mode: manual# 拒信不重新入队列default-requeue-rejected: falseretry:enabled: true #监听重试是否可用max-attempts: 5 #最大重试次数，默认为3retryTime: 120000 # 重试间隔时间（毫秒）

发送消息

创建一个RabbitMQ消息发送服务并注入到IoC容器中：

@Component
public class RabbitMQService {private final AmqpAdmin amqpAdmin;private final AmqpTemplate amqpTemplate;public RabbitMQService(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) {this.amqpAdmin = amqpAdmin;this.amqpTemplate = amqpTemplate;}public void sendMessage(String message){amqpTemplate.convertAndSend(message);}
}

消息接收

使用@RabbitListener注解接收来自于RabbitMQ的消息：

    @RabbitListener(queues = "someQueue")public void processMessage(String content) {log.info("recieve message from rabbitMQ:"+content);}

需要特别注意：

By default, if retries are disabled and the listener throws an exception, the delivery is retried indefinitely. You can modify this behavior in two ways: Set the defaultRequeueRejected property to false so that zero re-deliveries are attempted or throw an AmqpRejectAndDontRequeueException to signal the message should be rejected. The latter is the mechanism used when retries are enabled and the maximum number of delivery attempts is reached.

默认的重试被关闭的情况下，如果消息监听器抛出异常、发送方将会无限重试。我们可以通过如下两种方式避免这种情况：设置defaultRequeueRejected参数为false，因此失败消息不会重新进入队列。或者抛出AmqpRejectAndDontRequeueException异常表明消息应当被丢弃。第二种机制也是重试机制打开后、重试次数达到上限后采用的机制。

RabbitMQ涉及的内容比较多，详情后补。

春节愉快，开工大吉！

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

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

相关文章

深入理解Selenium：常用方法和操作详解

深入理解Selenium：常用方法和操作详解

前言 大家好&#xff0c;我是chowley&#xff0c;今天来复习一下Selenium的常用方法&#xff0c;并结合实战演练来加深记忆。 Selenium Selenium 是软件测试领域中最受欢迎的自动化测试工具之一&#xff0c;它提供了丰富的方法和操作来模拟用户在 Web 应用程序中的交互行为。…
阅读更多...
OpenAI发布首个视频生成模型Sora：输文字即可出视频现实还存在吗？

OpenAI发布首个视频生成模型Sora：输文字即可出视频现实还存在吗？

就在刚刚&#xff0c;openai发布了全新的视频生成人工智能模型Sora。用户只要通过文字输入一些提示语&#xff0c;就可以得到一个高清视频。同时Sora还可以根据静态图像生成相关的视频剪辑&#xff0c;效果相当炸裂。 这下又要干倒一大片创业公司了 Sora官方网址 Sora 目前…
阅读更多...
codeforces round 926 div2(A-D)

codeforces round 926 div2(A-D)

1.A a题 ∑ i 2 n ( a i − a i − 1 ) \sum_{i2}^{n}(a_{i}-a_{i-1}) ∑i2n​(ai​−ai−1​) a n − a 1 a_{n}-a_{1} an​−a1​所以我们排一下序输出 a n − a 1 a_{n}-a_{1} an​−a1​即可,当然直接累加也可以 #include<bits/stdc.h> using namespace std; #def…
阅读更多...
【矩阵】重塑矩阵

【矩阵】重塑矩阵

每日一道算法题之重塑矩阵 一、题目描述二、思路三、C代码 一、题目描述 题目来源&#xff1a;LeetCode 在 MATLAB 中&#xff0c;有一个非常有用的函数 reshape &#xff0c;它可以将一个 m x n 矩阵重塑为另一个大小不同&#xff08;p x q&#xff09;的新矩阵&#xff0c;但…
阅读更多...
（三十九）大数据实战——Prometheus监控平台的部署搭建

（三十九）大数据实战——Prometheus监控平台的部署搭建

前言 Prometheus监控&#xff08;Prometheus Monitoring&#xff09;是一种开源的系统监控和警报工具。它最初由SoundCloud开发并于2012年发布&#xff0c;并在2016年加入了云原生计算基金会&#xff08;CNCF&#xff09;。Prometheus监控旨在收集、存储和查询各种指标数据&am…
阅读更多...
17.3.2.7 二值化（内存处理）

17.3.2.7 二值化（内存处理）

版权声明&#xff1a;本文为博主原创文章&#xff0c;转载请在显著位置标明本文出处以及作者网名&#xff0c;未经作者允许不得用于商业目的。 二值化的算法请参看第17.3.1.5节。17.3.1.5 二值化&#xff08;黑白&#xff09;-CSDN博客 【例 17.48】二值化算法一。 //黑白1pr…
阅读更多...
GEE使用 Sentinel-1 SAR影像 和 Otsu 方法绘制洪水地图

GEE使用 Sentinel-1 SAR影像 和 Otsu 方法绘制洪水地图

洪水是世界上最常见、破坏性最大的自然灾害之一，造成了巨大的生命和财产损失。此外，随着气候变化的影响，近年来，洪灾变得更加频繁和不可预测。为了最大限度地减少生命和财产损失，必须迅速发现洪水蔓延的情况，并及时采取必要的干预措施。洪水蔓延探测大多使用光学传感器或…
阅读更多...
C++参悟：内存管理 shared_ptr

C++参悟：内存管理 shared_ptr

内存管理 shared_ptr 一、概述二、成员函数1. 构造函数2. 析构函数3. 修改器1. reset2. swap 4. 观察器1. get2. use_count3. operator bool 一、概述 std::shared_ptr 是通过指针保持对象共享所有权的智能指针。多个 shared_ptr 对象可占有同一对象。下列情况之一出现时销毁对…
阅读更多...
大模型基础架构入门

大模型基础架构入门

大模型架构 Prefix Decoder 和 Causal Decoder 和 Encoder-Decoder 区别 在于 attention mask不同&#xff1a; https://zhuanlan.zhihu.com/p/626310493 为何现在的大模型大部分是Decoder only结构&#xff1f; https://www.zhihu.com/question/588325646/answer/335725261…
阅读更多...
代码随想录算法训练营第四十四天 | 518. 零钱兑换 II、377. 组合总和 Ⅳ

代码随想录算法训练营第四十四天 | 518. 零钱兑换 II、377. 组合总和 Ⅳ

题目链接&#xff1a;518. 零钱兑换 II 文章讲解&#xff1a;代码随想录 518. 零钱兑换 II讲解 视频讲解&#xff1a;装满背包有多少种方法&#xff1f;组合与排列有讲究&#xff01;| LeetCode&#xff1a;518.零钱兑换II 思路和解法 题目&#xff1a; 给你一个整数数组 c…
阅读更多...
JVM-JVM中对象的结构

JVM-JVM中对象的结构

对象内存布局 对象里的三个区&#xff1a; 对象头&#xff08;Header&#xff09;&#xff1a;Java对象头占8byte。如果是数组则占12byte。因为JVM里数组size需要使用4byte存储。 标记字段MarkWord&#xff1a; 用于存储对象自身的运行时数据&#xff0c;它是synchronized实现轻…
阅读更多...
STM32——OLED菜单

STM32——OLED菜单

文章目录 一.补充二. 二级菜单代码 简介&#xff1a;首先在我的51 I2C里面有OLED详细讲解&#xff0c;本期代码从51OLED基础上移植过来的&#xff0c;可以先看完那篇文章&#xff0c;在看这个&#xff0c;然后按键我是用的定时器扫描不会堵塞程序,可以翻开我的文章有单独的定时…
阅读更多...
[2024]常用的conda指令

[2024]常用的conda指令

[2024]常用的conda指令 Hi&#xff0c;各位新年好&#xff0c;今天给大家介绍一下Python项目开发中常用的工具-Anaconda。 Anaconda指的是一个开源的Python发行版本&#xff0c;也是一个安装、管理Python相关包的软件。它自带了Python、Jupyter Notebook、Spyder等工具&#x…
阅读更多...
MIPS指令集处理器设计(支持64条汇编指令)

MIPS指令集处理器设计(支持64条汇编指令)

一、题目背景和意义 二、国内外研究现状 (略) 三、MIPS指令集处理器设计与实现 (一).MIPS指令集功能性梳理 1.MIPS指令集架构 (1).mips基础指令集格式总结 MIPS是&#xff08;Microcomputer without interlocked pipeline stages&#xff09;[10]的缩写&#xff0c;含义是…
阅读更多...
C++友元-＞全局函数做友元、类做友元、成员函数做友元

C++友元-＞全局函数做友元、类做友元、成员函数做友元

全局函数做友元代码&#xff1a; #include<iostream> using namespace std; #include<string> //建筑物类 class Building { //告诉编译器 goodGay全局函数 是 Building类的好朋友&#xff0c;可以访问类中的私有内容 friend void goodGay(Building * bu…
阅读更多...
第14讲投票帖子详情实现

第14讲投票帖子详情实现

投票帖子详情实现 后端,根据id查询投票帖子信息&#xff1a; /*** 根据id查询* param id* return*/ GetMapping("/{id}") public R findById(PathVariable(value "id")Integer id){Vote vote voteService.getById(id);WxUserInfo wxUserInfo wxUserInf…
阅读更多...
电商+支付双系统项目------设计数据库

电商+支付双系统项目------设计数据库

这篇文章将详细介绍电商支付双系统项目的数据库设计。数据库在该项目中扮演着至关重要的角色&#xff0c;它负责存储和管理用户信息、商品数据、订单记录以及支付交易等关键数据。通过精心设计和优化数据库结构&#xff0c;可以实现高效的数据存储和检索&#xff0c;确保系统的…
阅读更多...
JavaScript中的常见算法

JavaScript中的常见算法

一.排序算法 1.冒泡排序 冒泡排序比较所有相邻的两个项&#xff0c;如果第一个比第二个大&#xff0c;则交换它们。元素项向上移动至 正确的顺序&#xff0c;就好像气泡升至表面一样。 function bubbleSort(arr) {const { length } arrfor (let i 0; i < length - 1; i)…
阅读更多...
详解自定义类型：枚举与联合体！

详解自定义类型：枚举与联合体！

目录 ​编辑 一、枚举类型 1.枚举类型的声明 2.枚举类型的优点 3.枚举类型的使用 二、联合体类型(共用体&#xff09; 1.联合体类型的声明 2.联合体的特点 3.相同成员的结构体和联合体的对比 4.联合体大小的计算 5.用联合体判断大小端 三.完结散花 悟已往之不谏&…
阅读更多...
【Webpack】自动执行开发服务器 devServer

【Webpack】自动执行开发服务器 devServer

开发服务器&自动化 每次写完代码都需要手动输入指令才能编译代码&#xff0c;太麻烦了&#xff0c;我们希望一切自动化&#xff0c;即修改代码后服务器浏览器自动刷新。 1. 下载包 npm i webpack-dev-server -D2. 配置 webpack.config.js const path require("p…
阅读更多...
最新文章

Copyright @ 2022~2023