专栏首页王清培的专栏RabbitMQ 可靠投递

RabbitMQ 可靠投递

RabbitMQ 可靠投递

标签: RabbitMQ shovel-plugin ConfirmCallback RabbitMQ消息投递


  • 背景
  • confirmCallback 确认模式
  • returnCallback 未投递到 queue 退回模式
  • shovel-plugin 跨机房可靠投递

背景

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

rabbitmq 整个消息投递的路径为: producer->rabbitmq broker cluster->exchange->queue->consumer

messageproducerrabbitmq broker cluster 则会返回一个 confirmCallbackmessageexchange->queue 投递失败则会返回一个 returnCallback 。我们将利用这两个 callback 控制消息的最终一致性和部分纠错能力。

confirmCallback 确认模式

在创建 connectionFactory 的时候设置 PublisherConfirms(true) 选项,开启 confirmcallback

CachingConnectionFactory factory = new CachingConnectionFactory();
factory.setPublisherConfirms(true);//开启confirm模式
RabbitTemplate rabbitTemplate = new RabbitTemplate(factory);
rabbitTemplate.setConfirmCallback((data, ack, cause) -> {
        if (!ack) {
               log.error("消息发送失败!" + cause + data.toString());
        } else {
            log.info("消息发送成功,消息ID:" + (data != null ? data.getId() : null));
        }
    });

我们来看下 ConfirmCallback 接口。

public interface ConfirmCallback {

        /**
         * Confirmation callback.
         * @param correlationData correlation data for the callback.
         * @param ack true for ack, false for nack
         * @param cause An optional cause, for nack, when available, otherwise null.
         */
        void confirm(CorrelationData correlationData, boolean ack, String cause);

    }

重点是 CorrelationData 对象,每个发送的消息都需要配备一个 CorrelationData 相关数据对象,CorrelationData 对象内部只有一个 id 属性,用来表示当前消息唯一性。

发送的时候创建一个 CorrelationData 对象。

User user = new User();
user.setID(1010101L);
user.setUserName("plen");

rabbitTemplate.convertAndSend(exchange, routing, user,
        message -> {
        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
            return message;
        },
new CorrelationData(user.getID().toString()));

这里将 user ID 设置为当前消息 CorrelationData id 。当然这里是纯粹 demo,真实场景是需要做业务无关消息 ID 生成,同时要记录下这个 id 用来纠错和对账。

消息只要被 rabbitmq broker 接收到就会执行 confirmCallback,如果是 cluster 模式,需要所有 broker 接收到才会调用 confirmCallback

broker 接收到只能表示 message 已经到达服务器,并不能保证消息一定会被投递到目标 queue 里。所以需要用到接下来的 returnCallback

returnCallback 未投递到queue退回模式

confrim 模式只能保证消息到达 broker,不能保证消息准确投递到目标 queue 里。在有些业务场景下,我们需要保证消息一定要投递到目标 queue 里,此时就需要用到 return 退回模式。

同样创建 ConnectionFactory 到时候需要设置 PublisherReturns(true) 选项。

CachingConnectionFactory factory = new CachingConnectionFactory();
factory.setPublisherReturns(true);//开启return模式
rabbitTemplate.setMandatory(true);//开启强制委托模式

rabbitTemplate.setReturnCallback((message, replyCode, replyText,
                    exchange, routingKey) ->
    log.info(MessageFormat.format("消息发送ReturnCallback:{0},{1},{2},{3},{4},{5}", message, replyCode, replyText, exchange, routingKey)));

这样如果未能投递到目标 queue 里将调用 returnCallback ,可以记录下详细到投递数据,定期的巡检或者自动纠错都需要这些数据。

shovel-plugin 跨机房可靠投递

RabbitMQ 在跨机房集成提供了一个不错的插件 shovel 。使用 shovel-plugin 插件非常方便,shovel 可以接受机房之间的网络断开、机器下线等不稳定因素。

这里有两个 broker

10.211.55.3 rabbit_node1 10.211.55.4 rabbit_node2

我们希望将发送给 rabbit_node1 plen.queue 的消息传输到 rabbit_node2 plen.queue 中。我们先开启 rabbit_node1shovel-plugin

先看下当前 RabbitMQ 版本是否安装了 shovel-plugin,如果有的话直接开启。

rabbitmq-plugins  list
rabbitmq-plugins  enable rabbitmq_shovel
rabbitmq-plugins  enable rabbitmq_shovel_management

然后就可以在 Admin 面板里看到这个设置选项,怎么设置这里就不介绍了。主要就是配置下 amqp 协议地址,amqp://user:password@server-name/my-vhost

如果配置没有问题的话,应该是这样的一个状态,说明已经顺利连接到 rabbit_node2 broker

我们来看下 rabbit_node1rabbit_node2Connections 面板。 rabbit_node1(10.211.55.3):

rabbit_node2(10.211.55.4):

RabbitMQ shovel-plugin 插件在 rabbit_node1 broker 创建了两个 tcp 连接,端口 39544 连接是用来消费 plen.queue 里的消息,端口 55706 连接是用来推送消息给 rabbit_node2

我们来看下 rabbit_node1 tcp 连接状态:

tcp6       0      0 10.211.55.3:5672        10.211.55.3:39544       ESTABLISHED
tcp        0      0 10.211.55.3:55706       10.211.55.4:5672        ESTABLISHED

rabbit_node2 tcp 连接状态:

tcp6       0      0 10.211.55.4:5672        10.211.55.3:55706       ESTABLISHED

为了验证 shovel-plugin 稳定性,我们将 rabbit_node2 下线。

然后再发送消息,发现消息会现在 rabbit_node1 plen.queue 里待着,一旦 shovel-plugin 连接恢复将消费 rabbit_node1 plen.queue 消息,然后投递给 rabbit_node2 plen.queue

作者:王清培 (沪江集团资深JAVA架构师)

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • .NET深入解析LINQ框架(二:LINQ优雅的前奏)

    例子说明:假设我有一个表示学生的对象类型还有一个表示学生集合的类型。学生集合类型主要就是用来容纳学生实体,集合类型提供一系列的方法可以对这个集合进行连续的操作,...

    王清培
  • spring rest 容易被忽视的后端服务 chunked 性能问题

    spring boot 容易被忽视的后端服务 chunked 性能问题 标签(空格分隔): springboot springmvc chunked 背景 sp...

    王清培
  • zookeeper 实现分布式锁安全用法

    标签: zookeeper sessionExpire connectionLoss 分布式锁

    王清培
  • mysqldump命令详解 4-按条件备份表数据

    我们来备份test数据库中的innodb_table表和isam_table表中id为10的数据

    bsbforever
  • yum安装Ceph指定Jewel版本

    通过yum安装指定的rpm包,这个一般是 yum --showduplicates list ceph | expand,然后去通过yum安装指定的版本即可,这...

    用户2772802
  • pycharm使用笔记3-自动生成文件注释和函数注释

    版权声明:本文为博主原创文章,未经博主允许不得转载。python版本为python3,实例都是经过实际验证。 ...

    锦小年
  • 【程序源代码】Spring Shiro学习系统

    管理系统?基于springmvc、spring、mybatis-plus、shiro、easyui、Log4j2简单实用的权限系统。

    程序源代码
  • css重写checkbox样式

    一、前言 默认的checkbox长这样: <p> <span><input type="checkbox" /></sp...

    JMCui
  • Spring Boot从零入门2_核心模块详述和开发环境搭建

    在上一篇文章中我们已经详解介绍了Spring boot相关的概念、特性以及周边知识,这一篇文章我们继续深入。我们知道Spring boot是一个入口,利用Spr...

    别打名名
  • bootstrap label标签

    <span class="label label-default">默认标签</span> <span class="label label-primary"...

    用户5760343

扫码关注云+社区

领取腾讯云代金券