文档章节

RabbitMQ 可靠投递

Java烂猪皮
 Java烂猪皮
发布于 2018/08/16 20:36
字数 1184
阅读 7
收藏 0

RabbitMQ 可靠投递

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

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

背景

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

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

message 从 producer 到 rabbitmq broker cluster 则会返回一个 confirmCallback 。
message 从 exchange->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_node1 的 shovel-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_node1 和 rabbit_node2 的 Connections 面板。
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烂猪皮》公众号聊的不仅仅是Java技术知识,还有面试等干货,后期还有大量架构干货。大家一起关注吧!关注烂猪皮,你会了解的更多..............

本文转载自:https://www.cnblogs.com/wangiqngpei557/p/9381478.html

共有 人打赏支持
Java烂猪皮
粉丝 51
博文 30
码字总数 82549
作品 0
黄浦
私信 提问
网易蜂巢微服务架构:用RabbitMQ实现轻量级通信

微服务架构与MQ RabbitMQ场景分析与优化 RabbitMQ在网易蜂巢中的应用和案例分享 1微服务架构与MQ 微服务架构是一种架构模式,它将单体应用划分成一组微小的服务,各服务之间使用轻量级的通信...

andrewniu
2018/05/10
0
0
消息队列RabbitMQ入门介绍

(一)基本概念 RabbitMQ是流行的开源消息队列系统,用erlang语言开发。我曾经对这门语言挺有兴趣,学过一段时间,后来没坚持。RabbitMQ是AMQP(高级消息队列协议)的标准实现。如果不熟悉A...

icheer
2013/10/09
0
0
【转载】关于RabbitMQ的消息确认

RabbitMQ 将消息投递到客户端后,客户端如果没处理完这个消息就死掉了,这个消息还会不会存在?这取决于 RabbitMQ 的消息确认机制(Message acknowledgment)是否打开。 为了确保消息不会丢失...

摩云飞
2012/11/27
0
0
请教有关RabbitMQ消息确认的问题!

@摩云飞 你好,想跟你请教个问题: 下面是RabbitMQ的消息确认机制: “为了确保消息不会丢失,RabbitMQ支持消息确认机制。客户端在接受到消息并处理完后,可以发送一个ack消息给RabbitMQ,告...

庆沉
2014/01/02
8.8K
4
rabbitmq——prefetch count

消费者在开启acknowledge的情况下,对接收到的消息可以根据业务的需要异步对消息进行确认。 然而在实际使用过程中,由于消费者自身处理能力有限,从rabbitmq获取一定数量的消息后,希望rabbi...

hncscwc
2014/01/24
0
1

没有更多内容

加载失败,请刷新页面

加载更多

jdk11 HttpClient 爬虫

目的: 获得目标背单词网站中的单词, 写了一个简单的小爬虫, 使用jdk11 到此, 思路明确! 第一步, 把冰箱门...., 串词了,Sorry!! 第一步, 调用登陆接口, 拿到sessionid! 第二步, 带着sessionid...

GOToo
31分钟前
6
0
matlab-自控原理 taylor 泰勒展开 一、二元函数

  matlab : R2018a 64bit     OS : Windows 10 x64 typesetting : Markdown    blog : my.oschina.net/zhichengjiu    gitee : gitee.com/zhichengjiu   一元函数的泰勒展开 code c......

志成就
42分钟前
2
0
PreparedStatement批量执行sql

案例: 工具方法: public static Connection getConnection(){try {Class.forName("com.mysql.jdbc.Driver");ct = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/......

ZeroneLove
59分钟前
1
0
【scala】3.数组相关操作

简介 在本章中,我们将会学到如何在scala中操作数组。 1、定长数组 // 初始化长度为10的定长数组,每一个元素的值为0val nums = new Array[Int](10)// nums: Array[Int] = Array(0, 0, 0, ...

Areya
今天
2
0
教你零基础如何快速入门大数据技巧

现在是大数据时代,很多人都想要学习大数据,因为不管是就业前景还是薪资都非常的不错,不少人纷纷从其他行业转型到大数据行业,那么零基础的人也想要学习大数据怎么办呢?下面一起探讨下零基...

董黎明
今天
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部