RabbitMQ发布确认高级

news2024/11/16 6:34:48

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

发布确认

确认机制方案：

image

代码架构图：

image

配置文件

在application.properties文件中加入下面的配置：

spring.rabbitmq.publisher-confirm-type=correlated
  • NONE：禁用发布确认模式，是默认值
  • CORRELATED：发布消息成功到交换器后会触发回调方法
  • SIMPLE：有两种效果，其一效果和 CORRELATED 值一样会触发回调方法，其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑。

要注意的是waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker。

配置类

@Configuration
public class ConfirmConfig {
    public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";
    public static final String CONFIRM_QUEUE_NAME = "confirm.queue";

    /**
     * 声明交换机
     */
    @Bean("confirmExchange")
    public DirectExchange confirmExchange() {
        return new DirectExchange(CONFIRM_EXCHANGE_NAME);
    }

    /**
     * 声明队列
     * 该队列是确认队列
     */
    @Bean("confirmQueue")
    public Queue confirmQueue() {
        return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
    }

    /**
     * 声明交换机与队列绑定
     */
    @Bean
    public Binding queueBinding(@Qualifier("confirmQueue") Queue queue,
                                @Qualifier("confirmExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("key1");
    }
}

回调接口

@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback {
    /**
     * 交换机不管是否收到消息的一个回调方法
     *
     * @param correlationData 消息相关数据
     * @param ack             交换机是否收到消息
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            log.info("交换机已经收到 id 为：{}的消息", id);
        } else {
            log.info("交换机还未收到 id 为：{}消息，由于原因：{}", id, cause);
        }
    }
}

消息生产者

@Slf4j
@RestController
@RequestMapping("/confirm")
public class Producer {
    public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private MyCallBack myCallBack;

    /**
     * 依赖注入 rabbitTemplate 之后再设置它的回调对象
     */
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(myCallBack);
    }

    @GetMapping("sendMessage/{message}")
    public void sendMessage(@PathVariable String message) {
        // 指定消息 id 为 1
        CorrelationData correlationData1 = new CorrelationData("1");
        String routingKey = "key1";
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME, routingKey, message, correlationData1);
        log.info("发送消息routingKey={}，内容={}", routingKey, message);

        CorrelationData correlationData2 = new CorrelationData("2");
        routingKey = "key2";
        rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME, routingKey, message, correlationData2);
        log.info("发送消息routingKey={}，内容={}", routingKey, message);
    }
}

消息消费者

@Slf4j
@Component
public class ConfirmConsumer {

    public static final String CONFIRM_QUEUE_NAME = "confirm.queue";

    @RabbitListener(queues = CONFIRM_QUEUE_NAME)
    public void receiveMsg(Message message) {
        String msg = new String(message.getBody());
        log.info("接收到队列 confirm.queue 消息:{}", msg);
    }
}

测试

启动项目，发起一个请求：http://localhost:8080/confirm/sendMessage/xx1

发送了两条消息，第一条消息的 RoutingKey 为 “key1”，第二条消息的 RoutingKey 为 “key2”，两条消息都成功被交换机接收，也收到了交换机的确认回调，但消费者只收到了一条消息，因为第二条消息的 RoutingKey 与队列的 BindingKey 不一致，也没有其它队列能接收这个消息，所有第二条消息被直接丢弃了。

image

回退消息

在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，生产者是不知道消息被丢弃的。通过设置mandatory参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。

@Slf4j
@RestController
public class MessageProducer implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * rabbitTemplate 注入之后就设置该值
     */
    @PostConstruct
    private void init() {
        // 设置收到消息的回调交给谁处理
        rabbitTemplate.setConfirmCallback(this);
        // true：交换机无法将消息进行路由时，会将该消息返回给生产者
        // false：如果发现消息无法进行路由，则直接丢弃
        rabbitTemplate.setMandatory(true);
        // 设置回退消息交给谁处理
        rabbitTemplate.setReturnCallback(this);
    }

    @GetMapping("/sendMessage/{message}")
    public void sendMessage(@PathVariable String message) {
        // 让消息绑定一个 id 值
        CorrelationData correlationData1 = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("confirm.exchange", "key1", message + "key1", correlationData1);
        log.info("发送消息id为:{}内容为{}", correlationData1.getId(), message + "key1");

        CorrelationData correlationData2 = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("confirm.exchange", "key2", message + "key2", correlationData2);
        log.info("发送消息id为:{}内容为{}", correlationData2.getId(), message + "key2");
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            log.info("交换机收到消息确认成功，id:{}", id);
        } else {
            log.error("消息id:{}未成功投递到交换机，原因是:{}", id, cause);
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, 
            String exchange, String routingKey) {
        log.info("消息{}被服务器退回，退回原因:{}，交换机是:{}，路由 key:{}", new String(message.getBody()), replyText,
                exchange, routingKey);
    }
}

image

备份交换机

使用 mandatory 参数和回退消息，可以对无法投递消息的感知能力，在生产者的消息无法被投递时发现并处理。但有时候，并不知道该如何处理这些无法路由的消息，最多打个日志，然后触发报警，再来手动处理。

而通过日志来处理这些无法路由的消息是很不优雅的做法，特别是当生产者所在的服务有多台机器的时候，手动复制日志会更加麻烦而且容易出错。而且设置 mandatory 参数会增加生产者的复杂性，需要添加处理这些被退回的消息的逻辑。

如果既不想丢失消息，又不想增加生产者的复杂性，可以为队列设置死信交换机来存储处理失败的消息，可是这些不可路由消息根本没有机会进入到队列，因此无法使用死信队列来保存消息。

在 RabbitMQ 中，有一种备份交换机的机制存在，可以很好的应对这个问题。备份交换机可以理解为 RabbitMQ 中交换机的“备胎”，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout，这样就能把所有消息都投递到与其绑定的队列中，然后在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列。当然，还可以建立一个报警队列，用独立的消费者来进行监测和报警。

image

配置类

@Configuration
public class ConfirmConfig {

    public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";
    public static final String CONFIRM_QUEUE_NAME = "confirm.queue";
    public static final String BACKUP_EXCHANGE_NAME = "backup.exchange";
    public static final String BACKUP_QUEUE_NAME = "backup.queue";
    public static final String WARNING_QUEUE_NAME = "warning.queue";

    /**
     * 声明备份交换机
     */
    @Bean("backupExchange")
    public FanoutExchange backupExchange() {
        return new FanoutExchange(BACKUP_EXCHANGE_NAME);
    }

    /**
     * 声明警告队列
     */
    @Bean("warningQueue")
    public Queue warningQueue() {
        return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
    }

    /**
     * 声明警告队列和备份交换机绑定
     */
    @Bean
    public Binding warningBinding(@Qualifier("warningQueue") Queue queue,
                                  @Qualifier("backupExchange") FanoutExchange backupExchange) {
        return BindingBuilder.bind(queue).to(backupExchange);
    }

    /**
     * 声明备份队列
     */
    @Bean("backQueue")
    public Queue backQueue() {
        return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
    }

    /**
     * 声明备份队列和备份交换机绑定
     */
    @Bean
    public Binding backupBinding(@Qualifier("backQueue") Queue queue,
                                 @Qualifier("backupExchange") FanoutExchange backupExchange) {
        return BindingBuilder.bind(queue).to(backupExchange);
    }


    /**
     * 声明交换机
     */
    @Bean("confirmExchange")
    public DirectExchange confirmExchange() {
        ExchangeBuilder exchangeBuilder = ExchangeBuilder
                .directExchange(CONFIRM_EXCHANGE_NAME)
                .durable(true)
                // 设置该交换机的备份交换机
                .withArgument("alternate-exchange", BACKUP_EXCHANGE_NAME);
        return exchangeBuilder.build();
    }

    /**
     * 声明队列
     * 该队列是确认队列
     */
    @Bean("confirmQueue")
    public Queue confirmQueue() {
        return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
    }

    /**
     * 声明交换机与队列绑定
     */
    @Bean
    public Binding queueBinding(@Qualifier("confirmQueue") Queue queue,
                                @Qualifier("confirmExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("key1");
    }
}

消息生产者

和上面的保持不变

消息消费者

@Slf4j
@Component
public class ConfirmConsumer {

    public static final String CONFIRM_QUEUE_NAME = "confirm.queue";

    @RabbitListener(queues = CONFIRM_QUEUE_NAME)
    public void receiveMsg(Message message) {
        String msg = new String(message.getBody());
        log.info("接收到队列 confirm.queue 消息:{}", msg);
    }

    public static final String WARNING_QUEUE_NAME = "warning.queue";

    @RabbitListener(queues = WARNING_QUEUE_NAME)
    public void receiveWarningMsg(Message message) {
        String msg = new String(message.getBody());
        log.error("报警发现不可路由消息：{}", msg);
    }
}

测试

启动项目，发起一个请求：http://localhost:8080/sendMessage/aaaa

image

mandatory 参数与备份交换机可以一起使用的时候，如果两者同时开启，经过上面结果显示答案是备份交换机优先级高。

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

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

相关文章

Chrome安装油猴插件详细教程

Chrome安装油猴插件详细教程

Chrome安装油猴插件详细教程 一、油猴安装方法 方法一：Google官方商店安装（推荐，需要科学上网） 方法二：本地安装（无需科学上网，不会科学上网的适用） 二、安装油猴插件 方法一&am…
阅读更多...
FineReport数据图表制作-标签控件

FineReport数据图表制作-标签控件

1. 概述 1.1 版本 报表服务器版本 功能变更 11.0 -- 1.2 应用场景 「标签控件」不支持填报应用，只能在参数页面下使用。如下图所示： 注：移动端不识别「标签控件」，识别的是普通控件的「标签名称」属性，所以移动端…
阅读更多...
我不说你知道 DotImage SDK之DotViewer 在线文档查看编辑？

我不说你知道 DotImage SDK之DotViewer 在线文档查看编辑？

关注我 日日新，关注我 一手料，关注我 懂国外，关注我 更新快，关注我 不玩假！！！！ 开发团队被请求淹没，并被迫用更少的资源做更多的事情。DotViewer 旨在帮助解决这一挑战&…
阅读更多...
sequencer和sequence

sequencer和sequence

●了解了sequencer与driver之间传递sequence item的握手过程,同时也掌握了sequence与item之间的关系。 ●接下来需要就sequence挂载到sequencer的常用方法做出总结,大家可以通过对这些常用方法和宏的介绍，了解到它们不同的使用场景。 ●面对多个sequence如果需要同时…
阅读更多...
Python学习----静态web服务器

Python学习----静态web服务器

开发静态web服务器 开发步骤： 1、编写一个TCP服务端程序 2、获取浏览器发送的HTTP请求报文数据 3、读取固定页面数据，把页面数据组装HTTP响应报文数据发送给浏览器 4、HTTP响应报文数据发送完成之后，关闭服务于客户端的套接字 import socke…
阅读更多...
（Java）Mybatis学习笔记（四）

（Java）Mybatis学习笔记（四）

前言 继续学习自定义映射，今天便是mybatis初步学完的最后一天了，加油，奥里给~ 搭建MyBatis框架 步骤说明 创建表时把email打成了eamil，导致报了下错 1️⃣在mybatis下创建一个module 2️⃣配置pom.xml文件，导入相…
阅读更多...
[附源码]计算机毕业设计基于springboot的桌游信息管理系统

[附源码]计算机毕业设计基于springboot的桌游信息管理系统

项目运行 环境配置： Jdk1.8 Tomcat7.0 Mysql HBuilderX（Webstorm也行） Eclispe（IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持）。 项目技术： SSM mybatis Maven Vue 等等组成，B/S模式 M…
阅读更多...
深度解析布谷鸟过滤器（上篇）

深度解析布谷鸟过滤器（上篇）

深度解析布谷鸟过滤器 0 引言 布隆过滤器（Bloom Filter），诞生于UNIX元年（1970年）的一个老牛逼的过滤器，与时间戳同寿，经久不衰老而弥坚，查重性能至今令人非常满意。美中不足的是有…
阅读更多...
Windows任务计划程序Task Scheduler笔记

Windows任务计划程序Task Scheduler笔记

微软文档居然搜不到了 Windows任务计划程序已经存在许多年了，原来在微软的TechNet上有详细的操作介绍的，现在发现网站改版，原来的介绍居然搜索不到了，微软的平台上出现这种事情，也是比较吃惊了。 添加任务计划不难 …
阅读更多...
ORB-SLAM2 ---- Tracking::Relocalization函数

ORB-SLAM2 ---- Tracking::Relocalization函数

目录 1.函数作用 2.步骤 3.code 4.函数解释 4.1 将当前帧的描述子转化为BoW向量 4.2 用词袋找到与当前帧相似的候选关键帧 4.3 遍历所有的候选关键帧，通过词袋进行快速匹配，用匹配结果初始化PnP Solver 4.4 通过一系列操作，直到找到…
阅读更多...
项目成本管理软件能为你做什么？

项目成本管理软件能为你做什么？

成本管理与企业在当前以及未来项目中取得成功的能力密切相关。投资可靠的项目成本管理软件可以带来巨大的节约。一个好的成本管理解决方案不会把它当作一个孤立的功能，而是把它作为项目和投资组合绩效的关键因素加以利用，并在各个项目之间进行数据关联。…
阅读更多...
金山表单结果如何自动通知企业微信

金山表单结果如何自动通知企业微信

金山表单内置了丰富的模版，从表单、接龙、问卷、投票，可以满足你各种表单数据数据收集的需求。但是很多用户经常也会有一个痛点，通过金山表单收集的信息，如何才能实时通知企业微信/钉钉/飞书呢？ 比如防疫登记、安全复工…
阅读更多...
1.2 异步相关概念：深入了解

1.2 异步相关概念：深入了解

1.同步(Synchronous) VS 异步(Asynchronous) 所谓同步，可以理解为每当系统执行完一段代码或者函数后，系统将一直等待该段代码或函数返回的值或消息，直到系统接收到返回的值或消息后才继续往下执行下一段代码或者函数，在等待返回值…
阅读更多...
汽车行业大趋势——软件定义汽车

汽车行业大趋势——软件定义汽车

文章目录 前言一、软件定义汽车的驱动力二、SOA架构在软件定义汽车中的作用三、车载软件架构（内核、中间件、应用层）长期趋势总结前言 最早在2007年4月份的IEEE会议中提出“软件定义汽车”（SDV，Software Define Vehicle&#xff…
阅读更多...
flask-admin菜鸟学习笔记

flask-admin菜鸟学习笔记

近期在工作中需要维护若干个信息表，在这个过程中需要经常对表格进行操作、交叉操作、各个表格同步，和某平台信息同步。。。在此过程中需要建立一个“隐性”的流程，要第一步同步A和B，再同步B和C，。。。而检索更是痛苦&a…
阅读更多...
Python 中 4 个高效的技巧！

Python 中 4 个高效的技巧！

今天我想和大家分享 4 个省时的 Python 技巧，可以节省 10~20% 的 Python 执行时间。 反转列表 Python 中通常有两种反转列表的方法：切片或 reverse() 函数调用。这两种方法都可以反转列表，但需要注意的是内置函数 reverse() 会更改原始列表…
阅读更多...
最近面了12个人，发现这个测试基础题都答不上来...

最近面了12个人，发现这个测试基础题都答不上来...

一般面试我都会问一两道很基础的题目，来考察候选人的“地基”是否扎实，有些是操作系统层面的，有些是 python语言方面的，还有些… 深耕IT行业多年，我们发现，对于一个程序员而言，能去到一线互联网…
阅读更多...
托管海外服务器有哪些要求？

托管海外服务器有哪些要求？

对于成长中的企业来说，需要使用自己的Web 服务器来有效地控制网站运营安全以及关键数据管理。但与此同时，创建一个高效的数据中心需要巨大的基础设施成本，24小时现场经验丰富的技术和管理支持团队来管理所有数据中心需求、服务器和数据等等。…
阅读更多...
详细介绍NLP关键词提取算法

详细介绍NLP关键词提取算法

PageRank 算法 基于词图模型的关键词提取算法主要有 PageRank 和 TextRank。 PageRank 是 TextRank 算法的思想基础，TextRank 是 PageRank 在文本上的应用。 来源： Google 创始人拉里佩奇和谢尔盖布林于 1997 年构建早期的搜索系统原型时提出的链接分析…
阅读更多...
uniCloud云开发及一键生成代码

uniCloud云开发及一键生成代码

uniapp云开发（云数据库） 准备工作 一、新建项目选择云开发 关联云函数 在cloudfouctions右键点击创建云函数 在base下的index.js中写入 use strict; exports.main async (event, context) > {//event为客户端上传的参数console.log(event : , ev…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023