543、RabbitMQ详细入门教程系列 -【Confirm与Mandatory】 2022.09.05

一、前言概述

RabbitMQ原生系列文章RbbitMQ(五) – 100%消息投递消费中详细介绍过这两个操作

• Confirm：消息是否投递到Exchange中,具备两个回调函数,通过不同的回调函数判断是否成功。也就是说不管客户端是否将消息成功投递到Exchange都会触发这个监听回调
• Mandatory：消息是否投递到Queue中,当消息到达Exchange后因为内部错误、路由错误等原因导致消息未到达Queue都会触发该回调信息。回调的参数中就包括了丢失的消息

二、SpringAMQP配置Confirm

2.1 Connection设置

很好理解的一个事情,原生操作中如果需要添加Confirm回调第一步就是需要将channel设置为confirm模式通过APIchannel.confirmSelect()。SpringAMQP中的channel将通过配置的ConnectionFactory创建,自然需要提前进行声明。具体配置如下：将参数publisherConfirms参数设置为true

    <!--定义连接工厂-->
    <bean id="cacheConnectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
        <property name="addresses" value="${rabbitmq.host}:${rabbitmq.port}"/>
        <property name="username" value="${rabbitmq.username}"/>
        <property name="password" value="${rabbitmq.password}"/>
        <property name="connectionNameStrategy" ref="connectionNameStrategy"/>
        <!--声明信道为Confirm模式-->
        <property name="publisherConfirms" value="true"/>
    </bean>
1
2
3
4
5
6
7
8
9

2.2 template设置

完成Connection的配置以后还需要在template中配置confirm-callback属性,也就是你添加的Confirm处理实例

    <!--定义模板-->
    <rabbit:template id="jjRabbitTemplate" connection-factory="cacheConnectionFactory" message-converter="jackson2JsonMessageConvert"
  confirm-callback="jjConfirmCallBack"/>
1
2
3

2.3 Confirm实现

具体的Confirm实现就是实现org.springframework.amqp.rabbit.core.RabbitTemplate中的ConfirmCallback接口,当消息未发送到Exchange中时就会回调该实例中的confirm方法。具体参数含义作用如下表所示

/**
 * @version 1.0.0
 * @date: 2020/5/11 11:12
 **/
@Component
public class JjConfirmCallBack implements ConfirmCallback {
    
    @Override
    public void confirm (CorrelationData correlationData, boolean ack, String cause) {
        
    }
    
}
1
2
3
4
5
6
7
8
9
10
11
12
13

三、SpringAMQP配置ManDatory

3.1 XML配置

与Confirm配置一致,使用的时候都会要求对Connection和Template进行配置。SpringAMQP中的Mandatory被封装为ReturnCallBack,相对于Confirm来讲需要在Template中多配置一个参数mandatory。具体如下所示：

    <!--定义连接工厂-->
    <bean id="cacheConnectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
        <property name="addresses" value="${rabbitmq.host}:${rabbitmq.port}"/>
        <property name="username" value="${rabbitmq.username}"/>
        <property name="password" value="${rabbitmq.password}"/>
        <property name="connectionNameStrategy" ref="connectionNameStrategy"/>
        <property name="publisherConfirms" value="true"/>
        <!--声明ReturnCallBack-->
        <property name="publisherReturns" value="true"/>
    </bean>
1
2
3
4
5
6
7
8
9
10

    <!--定义模板-->
    <rabbit:template id="jjRabbitTemplate" connection-factory="cacheConnectionFactory" message-converter="jackson2JsonMessageConvert"
                     confirm-callback="jjConfirmCallBack" 
                     mandatory="true" return-callback="jjReturnCallBack"/>
1
2
3
4

3.2 ReturnCallBack实现

具体的ReturnCallBack实现与Confirm一致,都是去实现org.springframework.amqp.rabbit.core.RabbitTemplate中的ReturnCallback接口,参数列表含义如下：

/**
 * @author zsl
 * @version 1.0.0
 * @date: 2020/5/11 13:33
 **/
@Component
public class JjReturnCallBack implements ReturnCallback {
    
    @Override
    public void returnedMessage (Message message, int replyCode, String replyText, String exchange, String routingKey) {
    
    }
    
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

四、参考链接

[01] RabbitMQ详细入门教程系列 -【Confirm与Mandatory】