<?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 1. 设置 publisher‐confirms="true" ‐‐> <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" /> 

<!‐‐定义管理交换机、队列‐‐> 
<rabbit:admin connection‐factory="connectionFactory"/> 

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

<!‐‐2. 消息可靠性投递（生产端）‐‐> 
<rabbit:queue id="test_queue_confirm" name="test_queue_confirm"></rabbit:queue>     
   <rabbit:direct‐exchange name="test_exchange_confirm"> 
        <rabbit:bindings> 
            <rabbit:binding queue="test_queue_confirm" key="confirm"> 
            </rabbit:binding> 
        </rabbit:bindings> 
    </rabbit:direct‐exchange> 
</beans>

或者配置yml 文件

5, 编写测试代码

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

@Autowired 
private RabbitTemplate rabbitTemplate; 
/**
* 确认模式： 
* 步骤： 
* 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方法被执行了...."); 

        if (ack) { 
            //接收成功 
            System.out.println("接收成功消息" + cause); 
        } else { 
            //接收失败 
            System.out.println("接收失败消息" + cause); 
            //做一些处理，让消息再次发送。 
        } 
    } 
}); 

//3. 发送消息 
rabbitTemplate.convertAndSend("test_exchange_confirm111", "confirm", "message confirm...."); 
    } 
}

或

@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
public class confirmTest4 implements RabbitTemplate.ConfirmCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 确认模式：
     * 步骤：
     * 1，确认模式开启：在yml 配置文件里配置 publisher-confirms: true
     * 2, 在rabbitTemplate 定义ConfirmCallBack 回调函数
     */

    @Test
    public void testConfirms() {
        rabbitTemplate.setConfirmCallback(new confirmTest4());

        Map<String, String> msg = new HashMap<>();
        msg.put("pageId", "5abefd525b05aa293098fca6");
        //转成json串
        String jsonString = JSON.toJSONString(msg);
        rabbitTemplate.convertAndSend("ex_routing_cms_postpage", "5abefd525b05aa293098fca6", jsonString);
    }
    /**
     *
     * @param correlationData 相关配置信息
     * @param ack exchange交换机 是否成功收到了消息，true成功，false代表失败
     * @param cause 失败原因
     */

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("confirm 方法开始执行..............");

        if (ack) {
            //接收成功
            System.out.println("接收成功的消息：" + cause);
        } else {
            //接收失败
            System.out.println("接收失败消息:" + cause);
            //接收失败后，我们做一些处理，让消息再次发送，达到消息可靠性传递
        }
    }

其中重写confirm 方法里的参数，

? ? ? ? correlationData：消息唯一标识

? ? ? ? ack：确认结果

????????cause：失败原因

?这是发送消息到queue 成功后，展示的内容，

但是如果发送失败是这样的

?当发送失败是可以配置return 退回模式代码。

2，?return退回模式代码实现

回退模式：? 当消息发送给EXchange 后，Exchange 路由到Queue 失败是才会执行ReturnCallBack,具体实现如下：

? ? ? ? 1，在Spring-rabbitmq-producer.xml 配置文件，在rabbit:connection-factory 节点添加配置

publisher‐returns="true"

或者 在application.yml 中设置

设置交换机处理失败消息的模式，两种方法，一种写在代码里，一种配置文件

1， 配置文件

?2，代码方式：

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

    //设置交换机处理失败消息的模式
    rabbitTemplate.setMandatory(true);

    //2.设置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("return 执行了....");
        System.out.println(message);
        System.out.println(replyCode);
        System.out.println(replyText);
        System.out.println(exchange);
        System.out.println(routingKey);
        //处理
    }
});

    //3. 发送消息
    rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "message
    confirm....");
}

或

package com.xuecheng.manage_cms.returns;

import com.alibaba.fastjson.JSON;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.util.HashMap;
import java.util.Map;


@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
public class ReturnTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;



    @Test
    public void testReturn() {

        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("return 执行了..........");
                System.out.println("message = " + message);
                System.out.println("replyCode = " + replyCode);
                System.out.println("replyText = " + replyText);
                System.out.println("exchange = " + exchange);
                System.out.println("routingKey = " + routingKey);

            }
        });

        Map<String, String> msg = new HashMap<>();
        msg.put("pageId", "5abefd525b05aa293098fca6");
        //转成json串
        String jsonString = JSON.toJSONString(msg);
        rabbitTemplate.convertAndSend("ex_routing_cms_postpage", "5abefd525b05aa293098fca61", jsonString);
    }
}

设置routingKey为一个不符合规则的key,观察控制台打印结果。

returnedMessage方法中参数

????????消息主体message : message
? ? ? ? 消息主体 message : replyCode
? ? ? ? 描述：replyText
? ? ? ? 消息使用的交换器 exchange : exchange
? ? ? ? 消息使用的路由键 routing : routingKey

总结

对于确认模式

设置 ConnectionFactory 的 publisher-confifirms="true" 开启 确认模式。

使用 rabbitTemplate.setConfifirmCallback 设置回调函数。当消息发送到 exchange 后回调 confiirm方法。在方法中判断 ack ，如果为 true ，则发送成功，如果为 false ，则发送失败，需要处理。

<?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}"/>

<context:component‐scan base‐package="com.itheima.listener" />

<!‐‐定义监听器容器 添加 acknowledge="manual" 手动‐‐>
<rabbit:listener‐container connection‐factory="connectionFactory" acknowledge="manual"
>
<rabbit:listener ref="ackListener" queue‐names="test_queue_confirm">
</rabbit:listener>
</rabbit:listener‐container>
</beans>

/**
 * Consumer ACK机制：
 * 1. 设置手动签收。acknowledge="manual"
 * 2. 让监听器类实现ChannelAwareMessageListener接口
 * 3. 如果消息成功处理，则调用channel的 basicAck()签收
 * 4. 如果消息处理失败，则调用channel的basicNack()拒绝签收，broker重新发送给consumer
 */
@Component
public class ackListener implements ChannelAwareMessageListener {

    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "ex_routing_cms_postpage",type = "direct"),
            value = @Queue(value = "queue_cms_postpage_01",durable = "true"),
            key = "#.#"
    ))

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            //1, 接收转换消息
            System.out.println("接收消息内容："+new String(message.getBody()));
            //2, 处理业务逻辑
            System.out.println("处理业务逻辑");
            //int i = 2/0 ; // 代码运行错误
            //3,手动签收
            channel.basicAck(deliveryTag,true);
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("代码执行错误了。。。。。。");
            //4，代码出现错误后，设置拒绝签收，重新发送
            //第三个参数：requeue: 消息重回队列，如何设置为true，则消息重新回到queue,broker会重新发送该消息给消费端
            channel.basicNack(deliveryTag,true,true);
            //了解
            //channel.basicReject(deliveryTag,true);
        }
    }
}