11-RabbitMQ高级特性-消息可靠性投递

Devops海洋的渔夫

发布于 2022-11-22 09:56:54

3160

发布于 2022-11-22 09:56:54

文章被收录于专栏：Devops专栏

11-RabbitMQ高级特性-消息可靠性投递

消息的可靠投递

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

confirm 确认模式
return 退回模式

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

消息从 producer 到 exchange 则会返回一个 confirmCallback 。
消息从 exchange-->queue 投递失败则会返回一个 returnCallback 。

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

案例

1. confirm 确认模式

1.1 工程搭建

创建一个空的 maven 工程 rabbitmq-producer-spring：

1.2. 添加依赖

修改pom.xml文件内容为如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.lijw</groupId>
    <artifactId>rabbitmq-producer-spring</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>2.1.8.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>

        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>
    </dependencies>
</project>

1.3. 配置整合

创建rabbitmq.properties连接参数等配置文件；

rabbitmq.host=127.0.0.1
rabbitmq.port=5672
rabbitmq.username=libai
rabbitmq.password=libai
rabbitmq.virtual-host=/test

2.创建 spring-rabbitmq-producer.xml 整合配置文件；

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    <!--加载配置文件-->
    <context:property-placeholder location="classpath:rabbitmq.properties"/>

    <!-- 定义rabbitmq connectionFactory -->
    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"/>
    <!--定义管理交换机、队列-->
    <rabbit:admin connection-factory="connectionFactory"/>

    <!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
    
</beans>

上面是 rabbitmq 初始化的配置，下面我们来定义交换机与队列。

1.4 定义交换机与队列

<!--消息可靠性投递（生产端） -->
<!--  1.定义队列  -->
<rabbit:queue id="test_queue_confirm" name="test_queue_confirm"/>
<!--  2.定义交换机  -->
<rabbit:direct-exchange name="test_exchange_confirm">
    <!--  3. 绑定交换机与队列 -->
    <rabbit:bindings>
        <rabbit:binding queue="test_queue_confirm" key="confirm"></rabbit:binding>
    </rabbit:bindings>
</rabbit:direct-exchange>

1.5 开启确认模式

确认模式开启：ConnectionFactory中开启publisher-confirms="true"

1.6 编写测试方法，尝试发送消息

在rabbitTemplate定义ConfirmCallBack回调函数

package com.lijw;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

/**
 * @author Aron.li
 * @date 2022/3/4 20:41
 */
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 定义队列
    static final String QUEUE_NAME = "test_queue_confirm";
    // 定义交换机
    static final String EXCHANGE_NAME = "test_exchange_confirm";
    // 定义路由键
    static final String ROUTING_KEY = "confirm";

    /**
     * 确认模式：
     * 步骤：
     * 1. 确认模式开启：ConnectionFactory中开启publisher-confirms="true"
     * 2. 在rabbitTemplate定义ConfirmCallBack回调函数
     */
    @Test
    public void testConfirm() {
        // 2. 定义回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             *
             * @param correlationData 相关配置信息
             * @param ack             exchange交换机 是否成功收到了消息。true 成功，false代表失败
             * @param cause           失败原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("confirm方法被执行了");

                //通过ack判断是否发送消息成功
                if (ack) {
                    // 发送成功
                    System.out.println("ack: " + ack + ", 发送成功消息: " + cause);
                } else {
                    // 发送失败
                    System.out.println("ack: " + ack + ", 发送失败消息: " + cause);
                    // 错一些失败处理，让消息再次发送。
                }
            }
        });

        // 3. 发送消息
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, "test msg confirm 123.......");

        // 4. 阻止程序中断导致 channel 中断，如果 channel 中断，那么 confirm 返回的 ack 都只会是 false
        while (true) {

        }
    }

}

首先我们正常执行发送的方法，如下：

那么我们故意修改错误的交换机名称，让发送消息失败，如下：

以上就是 confirm 确认模式了。

2. return 退回模式

回退模式： 当消息发送给Exchange后，Exchange路由到Queue失败是 才会执行 ReturnCallBack

2.1. 开启回退模式:publisher-returns="true"

<!-- 定义rabbitmq connectionFactory -->
<rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                           port="${rabbitmq.port}"
                           username="${rabbitmq.username}"
                           password="${rabbitmq.password}"
                           virtual-host="${rabbitmq.virtual-host}"
                           publisher-confirms="true"
                           publisher-returns="true"
/>

2.2 编写测试方法，验证 returnCallBack 方法

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 定义队列
    static final String QUEUE_NAME = "test_queue_confirm";
    // 定义交换机
    static final String EXCHANGE_NAME = "test_exchange_confirm";
    // 定义路由键
    static final String ROUTING_KEY = "confirm";

    /**
     * 回退模式： 当消息发送给Exchange后，Exchange路由到Queue失败是 才会执行 ReturnCallBack
     * 步骤：
     * 1. 开启回退模式:publisher-returns="true"
     * 2. 设置ReturnCallBack
     * 3. 设置Exchange处理消息的模式：
     * 3.1. 如果消息没有路由到Queue，则丢弃消息（默认）
     * 3.2. 如果消息没有路由到Queue，返回给消息发送方ReturnCallBack
     */
    @Test
    public void testReturn() {

        // 1. 设置Exchange处理消息的模式
        rabbitTemplate.setMandatory(true); // 如果不设置，则默认丢弃消息

        // 2. 设置Exchange如果消息没有路由到Queue，返回给消息发送方ReturnCallBack
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             *
             * @param message     消息对象
             * @param replyCode   错误码
             * @param replyText   错误信息
             * @param exchange    交换机
             * @param routingKey  路由键
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("returnCallback 执行了...");

                System.out.println(message);
                System.out.println(replyCode);
                System.out.println(replyText);
                System.out.println(exchange);
                System.out.println(routingKey);

                // 错误处理... 后续篇章介绍
            }
        });

        // 3. 发送消息
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, "test msg returnCallBack 123.......");
        System.out.println("发送消息: test msg returnCallBack 123.......");

        // 4. 阻止程序中断导致 channel 中断，如果 channel 中断，那么 returnCallBack 将不会执行
        while (true) {

        }
    }

}

首先我们正常发送消息，如下：

那么下面我们来触发调用 returnCallback 方法，触发的方式是让消息达到 exchange，但是无法路由到 queue。也就是修改一个错误的 ROUTING_KEY 就可以了。

小结

设置ConnectionFactory的publisher-confirms="true" 开启确认模式。
使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。
设置ConnectionFactory的publisher-returns="true" 开启退回模式。
使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后，如果设置了rabbitTemplate.setMandatory(true)参数，则会将消息退回给producer。并执行回调函数returnedMessage。
在RabbitMQ中也提供了事务机制，但是性能较差，此处不做讲解。使用channel下列方法，完成事务控制：
- txSelect(), 用于将当前channel设置成transaction模式
- txCommit()，用于提交事务
- txRollback(),用于回滚事务

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2022-10-05，如有侵权请联系 cloudcommunity@tencent.com 删除

rabbitmq