RabbitMQ知识掌握 【进阶篇】

news2025/1/8 14:31:05

一、如何保证消息的可靠性 🍉

1.保证消息的可靠性投递 🥝

在生产环境中由于一些不明原因，导致 rabbitmq 重启，在 RabbitMQ 重启期间生产者消息投递失败，导致消息丢失，需要手动处理和恢复。于是，我们开始思考，如何才能进行 RabbitMQ 的消息可靠投递呢？特别是在这样比较极端的情况，RabbitMQ 集群不可用的时候，无法投递的消息该如何处理呢？

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。

消息从生产者到消费者的经历哪些组件:
生产者–交换机—队列—消费者

  • confirm 确认模式
  • return 退回模式
  • 消息从 producer 到 exchange 则会返回一个 confirmCallback 。
  • 消息从 exchange–>queue 投递失败则会返回一个 returnCallback 。

默认rabbitmq不开启上面两种模式。

我们将利用这两个 callback 控制消息的可靠性投递

在这里插入图片描述
confirm和return的实现

  1. 设置ConnectionFactory的publisher-confirm-type: correlated开启 确认模式。

  2. 使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。

  3. 设置ConnectionFactory的publisher-returns=“true” 开启 退回模式。

  4. 使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后执行回调函数returnedMessage。

confirm机制

（1）开启confirm

在这里插入图片描述

#开启confirm机制  默认为none会自动删除数据  开启手动模式 correlated
spring.rabbitmq.publisher-confirm-type=correlated

(2)设置rabbitTemplate的confirmCallback回调函数

在这里插入图片描述

  @Autowired
    private RabbitTemplate rabbitTemplate;

 @Test
    public void text01(){
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if (!b)
                System.out.println("系统繁忙请重新发送");
            }
        });

        rabbitTemplate.convertAndSend("zt_exchange","b.aaa", "你好，阿娇");
    }

return机制

在这里插入图片描述

#开启return机制  用来捕捉虚拟主机向队列中传递信息错误
spring.rabbitmq.publisher-returns=true

（2）设置回调

在这里插入图片描述

@Test
    public void text02(){
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                //当交换机发送消息到队列过程中失败启动当前方法
                System.out.println(replyCode);
            }
        });

        rabbitTemplate.convertAndSend("zt_exchange","b.v", "你好，阿娇");
    }

2.如何保证消息在队列中不丢失🥝

(1)设置队列为持久化

在这里插入图片描述

(2)设置消息的持久化

生产消息不设置过期时间默认为持久化

在这里插入图片描述

3.确保消息能可靠的被消费掉🥝

ACK确认机制

多个消费者同时收取消息，收取消息到一半，突然某个消费者挂掉，要保证此条消息不丢失，就需要acknowledgement机制，就是消费者消费完要通知服务端，服务端才将数据删除

这样就解决了，即使一个消费者出了问题，没有同步消息给服务端，还有其他的消费端去消费，保证了消息不丢的case。

ACK的实现

ack指Acknowledge，确认。 表示消费端收到消息后的确认方式。
有三种确认方式：
自动确认：acknowledge=“none”
手动确认：acknowledge=“manual”
根据异常情况确认：acknowledge=“auto”，（这种方式使用麻烦，并且不常用，不作讲解）

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息队列中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

在这里插入图片描述

#修改消费端  手动确认消息  默认自动确认 none
spring.rabbitmq.listener.simple.acknowledge-mode=manual

（2）修改代码

在这里插入图片描述

package com.lzq.listener;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
@Component//交于容器创建并管理
public class MyListener {
    @RabbitListener(queues={"zt_queue01"})
//queues：表示你监听的队列名
    public void hello01(Message message , Channel channel) throws IOException { //把监听到的消息封装到Mmessage类对象中

        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        byte[] body = message.getBody();
        String s= new String(body);
        System.out.println("123");
       // Map map = JSON.parseObject(s, HashMap.class);
        try {
            System.out.println("消息的内容:"+s);
            //basicAck:确认消息 -- rabbit服务端删除
            /*
            * long deliveryTag,第一个参数为消息的标志
            * boolean multiple 第二个参数为是否把该消费之前未确认的消息一起确认掉
            * */
            channel.basicAck(deliveryTag,true);
        }catch (Exception e){
            //basicNack:服务继续发送消息
            /*
            * long deliveryTag, boolean multiple, 前两个参数与上面的意义一样
            * boolean requeue 是否要求rabbitmq服务器重新发送该消息
            * */
            channel.basicNack(deliveryTag,true,true);
        }
//业务操作
    }
}

总结: 如何保证消息的可靠性?

[1] 保证消息的可靠性投递: confirm机制和return机制

[2] 队列中:—持久化

[3]使用ack机制保证消费者的可靠性消费。

二、延迟队列🍉

TTL🥝

TTL 全称 Time To Live（存活时间/过期时间）。

当消息到达存活时间后，还没有被消费，会被自动清除。

RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。

在这里插入图片描述

在这里插入图片描述
在这里插入图片描述
根据设置的时间十秒后会消失

 @Test
    public void test04(){

        /**
         * String exchange, String routingKey, Object message,
         * CorrelationData correlationData
         */
        Message message =new Message("咻咻咻".getBytes());
        message.getMessageProperties().setExpiration("10000");
        rabbitTemplate.send("bbb","a.b",message);
    }

小结:

设置队列过期时间使用参数：x-message-ttl，单位：ms(毫秒)，会对整个队列消息统一过期。

设置消息过期时间使用参数：expiration。单位：ms(毫秒)，当该消息在队列头部时（消费时），会单独判断这一消息是否过期。

如果两者都进行了设置，以时间短的为准。

死信队列🥝

死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是DLX。

在这里插入图片描述

什么样的消息会成为死信消息

  1. 队列消息长度到达限制；

  2. 消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；

  3. 原队列存在消息过期设置，消息到达超时时间未被消费；

队列绑定死信交换机：

在这里插入图片描述

在这里插入图片描述

延迟队列🥝

延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。

需求：

  1. 下单后，30分钟未支付，取消订单，回滚库存。

  2. 新用户注册成功7天后，发送短信问候。

实现方式：

  1. 定时器：性能差—每隔一段时间要进行数据库查询。

  2. 延迟队列

通过消息队列完成延迟队列的功能

很可惜，在RabbitMQ中并未提供延迟队列功能。

但是可以使用：TTL+死信队列 组合实现延迟队列的效果。

在这里插入图片描述

三、 如何防止消费者重复消费消息🍉

消息的幂等性—无论操作几次结果都是一样。

1、生成全局id，存入redis或者数据库，在消费者消费消息之前，查询一下该消息是否有消费过。

2、如果该消息已经消费过，则告诉mq消息已经消费，将该消息丢弃（手动ack）。

3、如果没有消费过，将该消息进行消费并将消费记录写进redis或者数据库中。

在这里插入图片描述

简单描述一下需求，如果订单完成之后，需要为用户累加积分，又需要保证积分不会重复累加。那么再mq消费消息之前，先去数据库查询该消息是否已经消费，如果已经消费那么直接丢弃消息。

生产者 🥝

package com.ykq.score.producer;
 
import com.alibaba.fastjson.JSONObject;
import com.xiaojie.score.entity.Score;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
 
import java.util.UUID;
 

@Component
@Slf4j
public class ScoreProducer implements RabbitTemplate.ConfirmCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    //定义交换机
    private static final String SCORE_EXCHANGE = "ykq_score_exchaneg";
    //定义路由键
    private static final String SCORE_ROUTINNGKEY = "score.add";
 
    /**
     * @description: 订单完成
     * @param:
     * @return: java.lang.String
     * @author xiaojie
     * @date: 2023/7/10 22:30
     */
    public String completeOrder() {
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单已完成");
        //发送积分通知
        Score score = new Score();
        score.setScore(100);
        score.setOrderId(orderId);
        String jsonMSg = JSONObject.toJSONString(score);
        sendScoreMsg(jsonMSg, orderId);
        return orderId;
    }
 

 
    @Async
    public void sendScoreMsg(String jsonMSg, String orderId) {
        this.rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.convertAndSend(SCORE_EXCHANGE, SCORE_ROUTINNGKEY, jsonMSg, message -> {
            //设置消息的id为唯一
            message.getMessageProperties().setMessageId(orderId);
            return message;
        });
    }
 
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String s) {
        if (ack) {
            log.info(">>>>>>>>消息发送成功:correlationData:{},ack:{},s:{}", correlationData, ack, s);
        } else {
            log.info(">>>>>>>消息发送失败{}", ack);
        }
    }
}

消费者🥝

package com.xiaojie.score.consumer;
 
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.xiaojie.score.entity.Score;
import com.xiaojie.score.mapper.ScoreMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
 
import java.io.IOException;
import java.util.Map;
 

@Component
@Slf4j
public class ScoreConsumer {
    @Autowired
    private ScoreMapper scoreMapper;
 
    @RabbitListener(queues = {"ykq_score_queue"})
    public void onMessage(Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {
        String orderId = message.getMessageProperties().getMessageId();
        if (StringUtils.isBlank(orderId)) {
            return;
        }
        log.info(">>>>>>>>消息id是：{}", orderId);
        String msg = new String(message.getBody());
        Score score = JSONObject.parseObject(msg, Score.class);
        if (score == null) {
            return;
        }
        //执行前去数据库查询，是否存在该数据，存在说明已经消费成功，不存在就去添加数据，添加成功丢弃消息
        Score dbScore = scoreMapper.selectByOrderId(orderId);
        if (dbScore != null) {
            //证明已经消费消息，告诉mq已经消费,丢弃消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }
        Integer result = scoreMapper.save(score);
        if (result > 0) {
            //积分已经累加，删除消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        } else {
            log.info("消费失败，采取相应的人工补偿");
        } 
    }
}

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

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

相关文章

微信小程序活动抽奖简单实现，包教包会

微信小程序活动抽奖简单实现，包教包会

问题&#xff1a; 针对用户使用活动抽奖&#xff0c;获得抽奖得到商品的成就以及满足感&#xff0c;那么我们应该怎么去实现小程序去转盘抽奖活动呢 例如&#xff1a;项目需要抽奖实现相应的奖品奖励 实现方法 实现的效果如下&#xff1a; 实现的主要代码&#xff1a; Page…
阅读更多...
ANSYS Workbench项目分析与案例实操详解

ANSYS Workbench项目分析与案例实操详解

&#x1f482; 个人网站:【海拥】【游戏大全】【神级源码资源网】&#x1f91f; 前端学习课程&#xff1a;&#x1f449;【28个案例趣学前端】【400个JS面试题】&#x1f485; 寻找学习交流、摸鱼划水的小伙伴&#xff0c;请点击【摸鱼学习交流群】 目录 前言准备工作创建项目导…
阅读更多...
盘点5款Cesium三维GIS渲染引擎

盘点5款Cesium三维GIS渲染引擎

目前&#xff0c;以Cesium为核心的各类产品繁多&#xff0c;本文将挑选一些以Cesium为核心的软件案例&#xff0c;为大家进行介绍。 1. CesiumJS CesiumJS相信凡是GIS行业相关人员都特别熟悉了&#xff0c;CesiumJS是一款开源的JavaScript库&#xff0c;用于创建高性能的地球…
阅读更多...
以太网（Ethernet）入门了解

以太网（Ethernet）入门了解

以太网&#xff08;Ethernet&#xff09;是一种常见的局域网&#xff08;LAN&#xff09;通信协议&#xff0c;它是由Xerox公司于1970年代中期开发的。以太网是一种基于广播技术的开放式网络协议&#xff0c;它允许设备在共享通信介质上进行通信。以下是关于以太网的基本概念、…
阅读更多...
如何做到有效的项目质量管理？这三点很关键

如何做到有效的项目质量管理？这三点很关键

“项目管理中&#xff0c;大家都遇到过的哪些绊脚石&#xff1f;” 近期看到了这样一个话题&#xff0c;大家纷纷吐槽了在项目中遇到的各种困难。除了常见的范围无限蔓延、进度不停延误、成本严重超支、内部沟通困难等一系列问题&#xff1b;其中提到次数较多的还有项目的质量…
阅读更多...
QOwnNotes Mac版-QOwnNotes for Mac(事务笔记管理)

QOwnNotes Mac版-QOwnNotes for Mac(事务笔记管理)

QOwnNotes for Mac是一款Mac平台上的笔记记录和待办事项软件&#xff0c;可以将你的笔记保存为纯文本文件&#xff0c;它支持 MarkDown 支持&#xff0c;并与 ownCloud 云服务紧密集成。 QOwnNotes 的亮点就是它集成了 ownCloud 云服务&#xff08;当然是可选的&#xff09;。…
阅读更多...
JavaWeb 速通XML

JavaWeb 速通XML

目录 一、XML快速入门 1.基本介绍 : 2.入门案例 : 二、XML语法 0.文件结构 : 1.文档声明 &#xff1a; 2. 元素 : 3.属性 : 4.注释 : 5.CDATA节 : PS : XML转义符 : 三、Dom4j 1.关于XML解析技术 : 2 Dom4j介绍 : 3.Dom4j使用 : 1 获取Document对象的三种方式 2 …
阅读更多...
ROS：Gazebo详解

ROS：Gazebo详解

目录 一、前言二、URDF与Gazebo基本集成流程2.1创建功能包2.2编写URDF文件2.3启动Gazebo并显示模型 三、URDF集成Gazebo相关设置3.1collision3.2inertial3.3颜色设置 四、URDF集成Gazebo实操4.1需求4.2流程4.3编写封装惯性矩阵算法的 xacro 文件4.4底盘 Xacro 文件4.5摄像头 Xa…
阅读更多...
5个值得收藏字体网站，优秀字体不用再找

5个值得收藏字体网站，优秀字体不用再找

如何在大量的字体网站中选择安全可靠的字体呢&#xff1f;这里本文收集整理了5个好用的字体网站&#xff0c;一起来看看吧&#xff01; 1、即时设计 对于设计师来说&#xff0c;一个好的字体网站要具备两个基本要素:美观和免费。如果你对字体有同样的要求&#xff0c;那么即时…
阅读更多...
给测试开发工程师的5条建议——你一定要知道

给测试开发工程师的5条建议——你一定要知道

目录 前言 质量心态 获取其他测试领域的知识 探索性测试 数据测试 可视化测试 可访问性测试 安全测试 混沌测试 获得正确的帮助 自动化不仅仅是执行测试脚本 分享也是学习的一种途径 最后 前言 近些年可以看出测试开发工程师是热度比较高的测试职位&#xff0c;除…
阅读更多...
【深度学习】多任务学习

【深度学习】多任务学习

多任务学习是一个很火的话题&#xff0c;在自动驾驶领域以及其他对存储资源要求较高的端侧任务&#xff0c;都会考虑这种方式。之前面试的时候也有被问题多任务学习。前几天看到一个写得很好的例子&#xff0c;在此记录一下&#xff0c;也好久没有写博客了。 1.参考 论文是20…
阅读更多...
samba服务

samba服务

samba服务 1. samba诞生背景2. samba简介3. samba配置文件详解3.1 配置文件中参数的作用3.1.1 security参数的四种验证方式3.1.2 passdb backend定义用户后台的类型 4. samba服务部署与配置windos共享linux挂载共享 1. samba诞生背景 1987 年&#xff0c;微软公司和英特尔公司…
阅读更多...
fastapi初使用，构建自己的api

fastapi初使用，构建自己的api

文章目录 1、安装2、api实现2.1、 app.get("/1")2.2、app.get("/{a}")2.3、app.get("/{a}{b}")2.4、函数和api分离 3、运行 原文链接&#xff1a;https://wangguo.site/posts/d98bb3c9.html fastapi 是一个基于 Python 的 API 构建框架&#xff…
阅读更多...
C国演义 [第八章]

C国演义 [第八章]

第八章 买卖股票的最佳时机题目理解步骤dp含义递推公式初始化遍历方向 代码 买卖股票的最佳时机II题目理解步骤dp含义递推公式初始化遍历方向 代码 买卖股票的最佳时机 力扣链接 给定一个数组 prices &#xff0c;它的第 i 个元素 prices[i] 表示一支给定股票第 i 天的价格 …
阅读更多...
IDEA使用方式

IDEA使用方式

1.翻译 1.Plugins插件&#xff1a;Chinese中文插件 文件F 编辑E 视图V 导航N 代码C 分析Z 重构R 构建B 运行U 工具T VCSS 窗口W 帮助H文件N 新建N 打开 打开最近 关闭项目 设置T 项目结构 文件属性 保存全部S 从磁盘全部重新加载 作废缓存/重启 导出/导入操作 其他设置 导出 打…
阅读更多...
路径规划算法：基于协作搜索优化的路径规划算法- 附代码

路径规划算法：基于协作搜索优化的路径规划算法- 附代码

路径规划算法&#xff1a;基于协作搜索优化的路径规划算法- 附代码 文章目录 路径规划算法&#xff1a;基于协作搜索优化的路径规划算法- 附代码1.算法原理1.1 环境设定1.2 约束条件1.3 适应度函数 2.算法结果3.MATLAB代码4.参考文献 摘要&#xff1a;本文主要介绍利用智能优化…
阅读更多...
suse linux 日语字库相关

suse linux 日语字库相关

问题 最近遇到一个日语字库的问题&#xff0c;界面上同一个label显示的日语字体大小都不一样&#xff0c; 如下图&#xff0c;圈出的部分字体大小和其他部分不一样。 分析 考虑两种情况 配置文件有问题缺少字库文件 检查配置文件 参考suse官方文档https://en.opensuse.o…
阅读更多...
5.带你入门matlab常见分布的期望和方差（matlab程序）

5.带你入门matlab常见分布的期望和方差（matlab程序）

代码及运行结果 %% 　二项分布的期望和方差 clear all; n1100; p10.3; [m1,v1]binostat(n1,p1) %100*0.3 100*0.3*0.7 %% %% 均匀分布的期望和方差 clear all; a11; b15; [m1,v1]unifstat(a1,b1) %% 正态分布的期望和方差 clear all; n12; n23; [m1,v1]normstat(n1,n2) %%…
阅读更多...
亚马逊真的赚钱吗？亚马逊多年卖家真实想法揭秘

亚马逊真的赚钱吗？亚马逊多年卖家真实想法揭秘

亚马逊跨境电商真的能赚钱吗&#xff1f;近期收到很多的咨询。理解大家对赚钱的渴望&#xff0c;但凡事都是有利有弊的。至于亚马逊跨境电商能不能赚钱&#xff0c;赚多少&#xff1f;咨询了很多亚马逊多年卖家&#xff0c;今天就跟大家一起聊聊吧&#xff01; 亚马逊跨境电商…
阅读更多...
解决dotnet调用https请求被中止未能创建SSL/TLS安全通道

解决dotnet调用https请求被中止未能创建SSL/TLS安全通道

环境:dotnet4.7.2/winserver2012 问题描述: 调用https出现请求被中止,未能创建SSL/TLS安全通道 第一试 自定义SetCertificatePolicy 函数&#xff0c;在建立http连接之前调用 SetCertificatePolicy 函数。 public static void SetCertificatePolicy() {ServicePointManage…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023