消息的可靠性投递

消息的可靠性投递是不仅仅是靠生产者端与消费者端的确认机制保证的，是由生产者端的发送消息的确认机制、消费者端的消息接收确认机制、重试机制与RabbitMQ持久化共同作用的。

一、准备工作

1.创建父工程

(1) 创建RabbitMQ-message 父工程

xml

<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.likebocai</groupId>
<artifactId>RabbitMQ-message</artifactId>
<version>1.0-SNAPSHOT</version>
<packaging>pom</packaging>
<modules>
    <module>message-producer</module>
    <module>message-consumer</module>
</modules>

<properties>
    <maven.compiler.source>17</maven.compiler.source>
    <maven.compiler.target>17</maven.compiler.target>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.0.5</version>
</parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
</dependencies>

二、生产者端消息确认

1.创建子模块 message-producer

pom.xml 不用写，继承父工程即可。

2.编写yaml文件

yaml
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: /
publisher-confirm-type: correlated
publisher-returns: true

publisher-confirm-type ：交换机确认功能的开启。

NONE 值是禁用发布确认模式，是默认值。
CORRELATED 值是发布消息成功到交换器后会触发回调方法，我们这次使用此值！
SIMPLE 值经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法，其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker;

publisher-returns ：队列确认功能的开启。

3.编写配置类

java
package com.likebocai.messageProducer.config;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;

@Configuration
@Slf4j
public class RabbitMQConfig implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {

// 增强 RabbitTemplate
@Resource
private RabbitTemplate rabbitTemplate;

@PostConstruct
public void initRabbitTemplate() {
    rabbitTemplate.setConfirmCallback(this);
    rabbitTemplate.setReturnsCallback(this);
}

@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    // 消息发送到交换机成功或失败，就调用这个方法！
    log.info("confirm() 的correlationData:" + correlationData);
    log.info("confirm() 的 ack:" + ack);
    log.info("confirm() 的 cause:" + cause);
}

@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
    // 消息发送到队列失败就调用这个方法！
    log.info("消息主体：" + new String(returnedMessage.getMessage().getBody()));
    log.info("应答码：" + returnedMessage.getReplyCode());
    log.info("描述：" + returnedMessage.getReplyText());
    log.info("消息使用的交换机 exchange：" + returnedMessage.getExchange());
    log.info("消息使用路由键 routing：" + returnedMessage.getRoutingKey());
}

}

(1) ConfirmCallback

重写此方法对应的则是配置yaml文件中的publisher-confirm-type: correlated

效果： 当消息发送到交换机成功或失败时就会叫用此方法！

(2) ReturnsCallback

重写此方法对应的则是配置yaml文件中的publisher-returns: true

效果：消息发送到队列失败就调用这个方法！

(3) 增强 RabbitTemplate

原本RabbitTemplate对象并没有生产者端的消息确认功能，要给它设置对应的组件才可以。

java
// 增强 RabbitTemplate
@Resource
private RabbitTemplate rabbitTemplate;

@PostConstruct
public void initRabbitTemplate() {
    rabbitTemplate.setConfirmCallback(this);
    rabbitTemplate.setReturnsCallback(this);
}

要点一设置RabbitTemplate

把RabbitTemplate 从IOC容器中取出，然后把回调函数的所在对象的设置到其中。

要点二@PostConstruct

@PostConstruct 官方文档

此注解的方法在整个Bean初始化的执行顺序是：

Constructor(构造方法)-->@Resource(依赖注入)-->@PostConstruct(注释的初始化方法)

使用此注解应该满足的条件：

• 该注解用于注释方法
• 如果注释的方法在拦截器(interceptor)类中必须满足以下两种格式之一：

​ 1.void <METHOD>(InvocationContext)

​ 2.Object <METHOD>(InvocationContext) throws Exception

• 如果注释的方法不是拦截器类，则必须格式为:

​ void <METHOD>() //不能有参数，返回值不为void， <METHOD>为方法名

• 注解的方法的访问修饰符可以是public, protected, package private or private

• 注释的方法可以用final修饰

• 注释的方法除了应用程序的客户端(application client)之外,不能是静态方法

• 如果该方法抛出未经检查的异常，则该类不得投入使用，除非 EJB 可以处理异常甚至从异常中恢复。

4.测试类发送消息验证结果

（1）编写测试类

java
package com.likebocai.messageProducer;

import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
public class test {
public static final String EXCHANGE_DIRECT = "exchange.direct.order";
public static final String ROUTING_KEY = "order";
public static final String QUEUE_NAME = "queue.order";

@Resource
private RabbitTemplate rabbitTemplate;

@Test
public void test01SendMessage() {
	rabbitTemplate.convertAndSend(EXCHANGE_DIRECT + "awd",ROUTING_KEY,"我是交换机出错！");
    rabbitTemplate.convertAndSend(EXCHANGE_DIRECT,ROUTING_KEY + "asd","我是找不到队列！");
}

}

（2）结果

1) 找不到队列

java
@Test
public void test01SendMessage() {
rabbitTemplate.convertAndSend(EXCHANGE_DIRECT,ROUTING_KEY + "asd","我是找不到队列！");
}

2) 找不到交换机

java
@Test
public void test01SendMessage() {
rabbitTemplate.convertAndSend(EXCHANGE_DIRECT + "awd",ROUTING_KEY,"我是交换机出错！");
}

输入结果：

2024-06-26T15:06:46.734+08:00 INFO 3956 --- [nectionFactory3] c.l.m.config.RabbitMQConfig : confirm() 的correlationData:null
2024-06-26T15:06:46.734+08:00 INFO 3956 --- [nectionFactory3] c.l.m.config.RabbitMQConfig : confirm() 的 ack:false
2024-06-26T15:06:46.734+08:00 INFO 3956 --- [nectionFactory3] c.l.m.config.RabbitMQConfig : confirm() 的 cause:channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'exchange.direct.orderawd' in vhost '/', class-id=60, method-id=40)

5.备份交换机

配置方法：

（1）添加备份交换机

注意备份交换机必须是广播模式！

（2）创建目标交换机绑定备份交换机

（3）为绑定目标交换机绑定消息队列

（4）为备份交换机创建、绑定队列

创建：

绑定：

（5）备份交换机测试结果

java
@Test
public void test01SendMessage() {
rabbitTemplate.convertAndSend(EXCHANGE_DIRECT,ROUTING_KEY,"我正常情况！");
rabbitTemplate.convertAndSend(EXCHANGE_DIRECT,ROUTING_KEY + "asd","我是找不到队列了！");
}

因为没有消费，所有他们两个队列都有消息！

三、消费者端消息确认

1.创建message-consumer子工程

（1）yaml

yaml
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: /
listener:
simple:
acknowledge-mode: manual

主要把消息确认模式改成手动确认！

（2）deliveyTag交付标签机制

（3）消费者代码

java
package com.likebocai.listener;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;

@Configuration
@Slf4j
public class MyRabbitMQListener {
public static final String QUEUE_NAME = "queue.order";
@RabbitListener(queues = {QUEUE_NAME})
public void processMessage(String dataString, Message message, Channel channel) throws IOException {
// 获取当前消息的deliveryTag
long deliveryTag = message.getMessageProperties().getDeliveryTag();

    try {
        // 核心代码
        log.info("消费端 内息内容：" + dataString);
        // 操作成功： 返回 ACK 信息
        // 第一个参数是 deliveryTag：代表此消息的唯一标识；第二个参数是 multiple：代表是否将此消息之前包括此消息进行批量操作 建议设置false 单独操作
        channel.basicAck(deliveryTag,false);
    } catch (Exception e) {
        // 获取当前消息是否有过重复性投递
        Boolean redelivered = message.getMessageProperties().getRedelivered();

        // 核心操作失败： 返回 NACK 信息
        // requeue 参数： 控制消息是否重新放回队列
        //      为 true 重新放回队列，broker 会重新投递这个消息
        //      为false 不重新放回队列，broker 会丢弃这个消息

        // 第一个参数是 deliveryTag：代表此消息的唯一标识；
        // 第二个参数是 multiple(false/true)：代表是否将此消息之前包括此消息进行批量操作 建议设置false 单独操作
        // 第三个参数是 requeue(false/true)： 表示是重新放回到队列中 按照实际情况选择
        if (redelivered) {
            // 此消息一定重复投递，不在放回队列
            channel.basicNack(deliveryTag,false,false);
        } else {
            // 说明该消息是第一次投递 ，失败了可以再头第一次
            channel.basicNack(deliveryTag,false,true);
        }

        throw new RuntimeException(e);
    }
}

}

此代码的重点是Message和Channel。

Message：是该消息的主要信息，包括消息的deliveryTag和消息是否重复投递的标识redelivered

Channel：主要是操作broker是否对消息进行关闭与重复投递。

（5）测试结果

测试结果是没什么好测试的。

四、RabbitMQ持久化

RabbitMQ默认持久化！

完结撒花！！！