SpringBoot中的RabbitMQ用法詳解

更新時間：2023年07月02日 11:37:09 作者：程序媛-徐師姐

RabbitMQ是一個開源的消息隊列系統(tǒng)，它通過AMQP（高級消息隊列協(xié)議）來實現(xiàn)消息的傳遞，SpringBoot是目前非常流行的Java開發(fā)框架，它提供了很多便利性的功能，其中就包括對RabbitMQ的支持，在本文中，我們將介紹如何在SpringBoot中使用RabbitMQ

Spring Boot中的 RabbitMQ 是什么，如何使用

RabbitMQ 的基本概念

在使用 RabbitMQ 之前，我們需要了解一些基本概念。

消息隊列

消息隊列是一種用于異步通信的機制。消息發(fā)送者將消息發(fā)送到隊列中，消息接收者從隊列中獲取消息。通過消息隊列，可以實現(xiàn)消息的異步傳遞，降低系統(tǒng)之間的耦合性。

消息

消息是指需要傳遞的數(shù)據(jù)。

生產(chǎn)者

生產(chǎn)者是指向消息隊列發(fā)送消息的程序。

消費者

消費者是指從消息隊列中獲取消息并處理的程序。

隊列

隊列是指消息存儲的地方。

交換機

交換機是用于接收生產(chǎn)者發(fā)送的消息，并將消息路由到相應(yīng)的隊列中。

路由鍵

路由鍵是一個字符串，用于指定消息應(yīng)該被路由到哪個隊列中。

綁定

綁定是指將隊列和交換機連接起來的過程。

如何在 Spring Boot 中使用 RabbitMQ

添加依賴

首先，我們需要在 Maven 或 Gradle 中添加 RabbitMQ 的依賴。在 Maven 中，我們可以添加以下依賴：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

在 Gradle 中，我們可以添加以下依賴：

implementation 'org.springframework.boot:spring-boot-starter-amqp'

配置 RabbitMQ

在 Spring Boot 中，我們可以使用 application.yml 或者 application.properties 文件來配置 RabbitMQ。

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

創(chuàng)建生產(chǎn)者

下面是一個簡單的 RabbitMQ 生產(chǎn)者示例：

@Component
public class RabbitMQProducer {
    private final RabbitTemplate rabbitTemplate;
    public RabbitMQProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    public void send(String message) {
        rabbitTemplate.convertAndSend("myExchange", "myRoutingKey", message);
    }
}

在這個示例中，我們使用了 Spring AMQP 提供的 RabbitTemplate 類來發(fā)送消息。convertAndSend 方法用于將消息發(fā)送到指定的交換機和路由鍵。

創(chuàng)建消費者

下面是一個簡單的 RabbitMQ 消費者示例：

@Component
public class RabbitMQConsumer {
    @RabbitListener(queues = "myQueue")
    public void receive(String message) {
        System.out.println("Received message: " + message);
    }
}

在這個示例中，我們使用了 Spring AMQP 提供的 @RabbitListener 注解來指定消費者應(yīng)該監(jiān)聽哪個隊列。當有消息到達隊列時，receive 方法會被調(diào)用，并且接收到消息的內(nèi)容會作為參數(shù)傳遞給該方法。

運行示例

現(xiàn)在我們已經(jīng)創(chuàng)建了一個簡單的 RabbitMQ 應(yīng)用程序。我們可以在 main 方法中創(chuàng)建 Spring Boot 應(yīng)用程序，并在其中注入我們的生產(chǎn)者和消費者。然后，我們可以使用生產(chǎn)者向隊列發(fā)送消息，消費者將會接收到這些消息并輸出到控制臺。

@SpringBootApplication
public class Application implements CommandLineRunner {
    private final RabbitMQProducer rabbitMQProducer;
    private final RabbitMQConsumer rabbitMQConsumer;
    public Application(RabbitMQProducer rabbitMQProducer, RabbitMQConsumer rabbitMQConsumer) {
        this.rabbitMQProducer = rabbitMQProducer;
        this.rabbitMQConsumer = rabbitMQConsumer;
    }
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
    @Override
    public void run(String... args) throws Exception {
        rabbitMQProducer.send("Hello, RabbitMQ!");
    }
}

運行程序后，我們可以在控制臺看到類似于以下輸出：

Received message: Hello, RabbitMQ!

這表明我們已經(jīng)成功地向隊列發(fā)送了一條消息，并且消費者已經(jīng)成功地接收到了這個消息。

RabbitMQ 高級功能

除了基本功能之外，RabbitMQ 還提供了一些高級功能，例如：

消息確認

當生產(chǎn)者發(fā)送一條消息時，它并不知道這條消息是否已經(jīng)被成功處理。如果消息沒有被成功處理，生產(chǎn)者將會不斷地嘗試重發(fā)這條消息，直到它被成功處理為止。

為了解決這個問題，RabbitMQ 提供了消息確認機制。當生產(chǎn)者發(fā)送一條消息時，它可以請求 RabbitMQ 確認這條消息是否已經(jīng)被成功處理。如果消息已經(jīng)被成功處理，RabbitMQ 將會發(fā)送一個確認消息給生產(chǎn)者。

@Component
public class RabbitMQProducer {
    private final RabbitTemplate rabbitTemplate;
    public RabbitMQProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    public void send(String message) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("myExchange", "myRoutingKey", message, correlationData);
    }
    @Bean
    public ConfirmCallback confirmCallback() {
        return (correlationData, ack, cause) -> {
            if (ack) {
                System.out.println("Message with correlation id " + correlationData.getId() + " has been confirmed");
            } else {
                System.out.println("Message with correlation id " + correlationData.getId() + " has been rejected: " + cause);
            }
        };
    }
}

在這個示例中，我們在生產(chǎn)者中使用了 CorrelationData 類來跟蹤消息。我們還創(chuàng)建了一個 ConfirmCallback bean 來處理消息確認。當消息被成功處理時，confirmCallback 方法將會被調(diào)用，并輸出一條確認消息。當消息被拒絕時，confirmCallback 方法也將會被調(diào)用，并輸出一條拒絕消息。

消息持久化

默認情況下，RabbitMQ 不會將消息持久化到磁盤上。如果 RabbitMQ 在崩潰之前沒有將消息發(fā)送給消費者，這些消息將會丟失。

為了解決這個問題，我們可以將消息標記為持久化。這樣，即使 RabbitMQ 崩潰，消息也會被保存到磁盤上，并在 RabbitMQ 重啟后重新發(fā)送給消費者。

@Component
public class RabbitMQProducer {
    private final RabbitTemplate rabbitTemplate;
    public RabbitMQProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    public void send(String message) {
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        Message messageObject = new Message(message.getBytes(), messageProperties);
        rabbitTemplate.send("myExchange", "myRoutingKey", messageObject);
    }
}

在這個示例中，我們在生產(chǎn)者中使用了 MessageProperties 類來設(shè)置消息的持久化屬性。我們還使用了 MessageDeliveryMode.PERSISTENT 枚舉值來標記消息為持久化消息。

消息 TTL

消息 TTL（Time To Live）是指消息在隊列中存儲的時間。如果消息在指定的時間內(nèi)沒有被消費者消費，它將會被從隊列中自動刪除。

@Component
public class RabbitMQProducer {
    private final RabbitTemplate rabbitTemplate;
    public RabbitMQProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    public void send(String message) {
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setExpiration("5000"); // 5 seconds
        Message messageObject = new Message(message.getBytes(), messageProperties);
        rabbitTemplate.send("myExchange", "myRoutingKey", messageObject);
    }
}

在這個示例中，我們在生產(chǎn)者中使用了 MessageProperties 類來設(shè)置消息的 TTL 屬性。我們將 messageProperties.setExpiration("5000") 設(shè)置為 5000 毫秒，這意味著消息在隊列中最多存儲 5 秒鐘。

死信隊列

死信隊列是指當消息被拒絕或者過期時，它將會被重新路由到另一個隊列中。這個隊列就被稱為死信隊列。

@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue myQueue() {
        return QueueBuilder.durable("myQueue")
                .withArgument("x-dead-letter-exchange", "myDeadLetterExchange")
                .withArgument("x-dead-letter-routing-key", "myDeadLetterRoutingKey")
                .build();
    }
    @Bean
    public Queue myDeadLetterQueue() {
        return QueueBuilder.durable("myDeadLetterQueue").build();
    }
    @Bean
    public Exchange myExchange() {
        return ExchangeBuilder.directExchange("myExchange").durable(true).build();
    }
    @Bean
    public Exchange myDeadLetterExchange() {
        return ExchangeBuilder.directExchange("myDeadLetterExchange").durable(true).build();
    }
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(myQueue()).to(myExchange()).with("myRoutingKey").noargs();
    }
    @Bean
    public Binding deadLetterBinding() {
        return BindingBuilder.bind(myDeadLetterQueue()).to(myDeadLetterExchange()).with("myDeadLetterRoutingKey").noargs();
    }
}

在這個示例中，我們創(chuàng)建了一個名為 myQueue 的隊列，并使用 withArgument 方法來指定它的死信交換機和路由鍵。我們還創(chuàng)建了一個名為 myDeadLetterQueue 的隊列，并將其綁定到名為 myDeadLetterExchange 的交換機上。最后，我們創(chuàng)建了綁定，將 myQueue 隊列綁定到 myExchange 交換機上。

當消息在 myQueue 中被拒絕或過期時，它將會被重新路由到 myDeadLetterExchange 交換機，并將其路由到 myDeadLetterQueue 隊列中。

總結(jié)

本文介紹了 RabbitMQ 的基本概念，以及如何在 Spring Boot 中使用 RabbitMQ。我們還介紹了 RabbitMQ 的一些高級功能，包括消息確認、消息持久化、消息 TTL 和死信隊列。通過學(xué)習(xí)本文，你應(yīng)該已經(jīng)有了足夠的知識來開始在 Spring Boot 中使用 RabbitMQ 了。

以上就是SpringBoot中的RabbitMQ的用法詳解的詳細內(nèi)容，更多關(guān)于SpringBoot RabbitMQ的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: