RabbitMQ（六）消息的持久化

本文介绍: RabbitMQ（六）消息的持久化

在这里插入图片描述

出现消息丢失的场景有：

直接实例化对应的 Exchange 实现类即可，默认：持久化的、非自动删除的。

@Bean
public DirectExchange directExchange() {
    return new DirectExchange(directExchange);
}

Exchange 源码：

package org.springframework.amqp.core;

public abstract class AbstractExchange extends AbstractDeclarable implements Exchange {
    ...
    public AbstractExchange(String name) {
        // 默认是持久化的、非自动删除的。
        this(name, true, false);
    }
    ...
}

在 RabbitMQ 的原生 API 中，例如：Java 客户端 API，声明持久化交换机时，需要将 durable 参数设置为 true：

// 声明交换机：(交换机名称，交换机类型，持久化)
channel.exchangeDeclare("myExchange", "direct", true);

在 Spring AMQP 中，我们可以使用 ExchangeBuilder 来创建持久化的交换机：

import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;

Exchange exchange = ExchangeBuilder.directExchange("myExchange")
    							.durable(true) // 交换机持久化
        						.build();

直接实例化 Queue 类即可，默认：持久化的、非独占的、非自动删除的。

@Bean
public Queue myQueue() {
    return new Queue("myQueue");
}

Queue 源码：

package org.springframework.amqp.core;

public Queue(String name) {
    // 默认是持久化的、非独占的、非自动删除的。
   this(name, true, false, false);
}

在原生 RabbitMQ API 中，例如 Java 客户端 API，声明持久化队列时，需要将 durable 参数设置为 true：

// 声明队列：(队列名称，持久化，非独占，非自动删除，可选队列参数)
channel.queueDeclare("myQueue", true, false, false, null);

在 Spring AMQP 中，可以使用 QueueBuilder 来创建持久化的队列：

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;

Queue queue = QueueBuilder.durable("myQueue").build();

在原生 RabbitMQ API 中，例如 Java 客户端 API，声明持久化消息时，需要将 deliveryMode 参数设置为 2：

import com.rabbitmq.client.AMQP;

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2) // 这里2对应MessageDeliveryMode.PERSISTENT
    .build();

// 参数：(交换机，路由键，消息的其他参数，消息体)
channel.basicPublish("myExchange", "myRoutingKey", props. messageBytes);

或者，我们可以直接使用 MessageProperties.PERSISTENT_TEXT_PLAIN 来进行指定消息的持久化：

import com.rabbitmq.client.MessageProperties;

// 参数：(交换机，路由键，消息的其他参数，消息体)
channel.basicPublish("myExchange", "myRoutingKey", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

在 Spring AMQP 中，可以使用 rabbitTemplate.convertAndSend() 来创建持久化的消息：

rabbitTemplate.convertAndSend("myQueue", body, message -> {
    message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
    return message;
});