当前位置：首页 > news >正文

Springboot整合rabbitmq并实现消息可靠性和持久性

news 来源：原创 2024/4/28 4:29:48

Springboot整合rabbitmq并实现消息可靠性和持久性

1. 环境准备
2. 创建 Spring Boot 项目
3. 配置 RabbitMQ
4. 实现消息的可靠性和持久性
- 4.1 发送可靠的消息
- 4.2 接收可靠的消息
5. 运行应用程序
6. 总结

Spring Boot 是一种快速构建应用程序的框架，而 RabbitMQ 是一种消息代理，它提供了可靠的消息传递服务。将 Spring Boot 与 RabbitMQ 结合使用，可以方便地实现消息传递，同时确保消息的可靠性和持久性。

在这里插入图片描述

本文将演示如何使用 Spring Boot 整合 RabbitMQ，并实现消息的可靠性和持久性。

1. 环境准备

在开始之前，您需要安装以下软件：

JDK 1.8 或更高版本
Maven 3.0 或更高版本
RabbitMQ 3.5.0 或更高版本

2. 创建 Spring Boot 项目

首先，我们需要创建一个 Spring Boot 项目。您可以使用 Spring Initializr 快速创建项目。

创建项目时，请确保选择以下依赖项：

Spring Boot DevTools
Spring Web
Spring AMQP

这些依赖项将帮助我们创建一个 Spring Boot 应用程序，并使用 Spring AMQP 库连接到 RabbitMQ。

3. 配置 RabbitMQ

在连接到 RabbitMQ 之前，我们需要配置 RabbitMQ。您可以使用以下命令安装 RabbitMQ：

brew install rabbitmq

安装完成后，启动 RabbitMQ 服务：

brew services start rabbitmq

然后，打开浏览器，访问 http://localhost:15672，您将看到 RabbitMQ 的管理控制台。在这里，您可以创建队列，交换机等。

现在，我们需要创建一个队列和交换机。我们将使用 rabbitTemplate 来发送消息和接收消息，所以我们需要确保队列和交换机已经创建。

您可以使用以下代码创建队列和交换机：

@Configuration
public class RabbitConfig {
    public static final String QUEUE_NAME = "my-queue";
    public static final String EXCHANGE_NAME = "my-exchange";
    public static final String ROUTING_KEY = "my-routing-key";

    @Bean
    public Queue queue() {
        return new Queue(QUEUE_NAME);
    }

    @Bean
    public DirectExchange exchange() {
        return new DirectExchange(EXCHANGE_NAME);
    }

    @Bean
    public Binding binding(Queue queue, DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }
}

这将创建一个名为 my-queue 的队列和一个名为 my-exchange 的交换机，并将它们绑定在一起，使用 my-routing-key 作为路由键。

4. 实现消息的可靠性和持久性

现在，我们已经设置好了 RabbitMQ，并且有了一个队列和一个交换机，我们可以开始实现消息的可靠性和持久性了。

4.1 发送可靠的消息

要发送可靠的消息，我们需要确保消息已经成功到达 RabbitMQ，并且已经被持久化。为此，我们可以使用以下代码：

@Service
public class MessageSender {
    private final RabbitTemplate rabbitTemplate;
	public void sendMessage(String message) {
	    rabbitTemplate.setMandatory(true);
	    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
	        if (!ack) {
	            // 处理消息发送失败的情况
	        }
	    });
		// 构建消息体
	    Message msg = MessageBuilder.withBody(message.getBytes())
	            .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
	            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
	            .build();
		// 发送MQ消息
	    rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME, RabbitConfig.ROUTING_KEY, msg);
	}
}

在这里，我们使用 rabbitTemplate 来发送消息，并启用了 mandatory 选项，这将确保消息被成功发送到 RabbitMQ。

我们还设置了 confirmCallback 回调函数，用于处理消息发送失败的情况。如果消息无法被正确发送，我们可以在这个回调函数中执行处理逻辑。

此外，我们还设置了消息的传输模式为持久模式，这将确保消息在 RabbitMQ 挂掉时不会丢失。

4.2 接收可靠的消息

为了接收可靠的消息，我们需要在 RabbitMQ 中创建一个消费者，并启用手动确认模式。手动确认模式将允许我们在处理消息之后，手动向 RabbitMQ 确认该消息已被处理。

以下是一个简单的消费者示例：

@Service
public class MessageReceiver {
    @RabbitListener(queues = RabbitConfig.QUEUE_NAME)
    public void handleMessage(Message message, Channel channel) throws Exception {
        String msg = new String(message.getBody(), StandardCharsets.UTF_8);
        System.out.println("Received message: " + msg);

        try {
            // 处理消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            // 处理消息处理失败的情况
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}