一文掌握Springboot集成RabbitMQ的方法

更新時(shí)間：2024年04月02日 10:53:01 作者：有一只柴犬

RabbitMQ是開(kāi)源的消息隊(duì)列服務(wù)器,實(shí)現(xiàn)了AMQP高級(jí)消息隊(duì)列協(xié)議標(biāo)準(zhǔn),AMQP?是一種應(yīng)用層協(xié)議,為面向消息的中間件設(shè)計(jì),基于此協(xié)議的客戶端與消息中間件可傳遞消息,并不受產(chǎn)品、開(kāi)發(fā)語(yǔ)言等條件的限制,這篇文章主要介紹了Springboot集成RabbitMQ的過(guò)程,需要的朋友可以參考下

1、前言

消息隊(duì)列（Message Queue，簡(jiǎn)稱(chēng) MQ）是一種異步的消息傳遞中間件，它解耦了應(yīng)用程序之間的通信。應(yīng)用程序可以將消息發(fā)送到隊(duì)列，而無(wú)需知道誰(shuí)會(huì)接收這些消息。接收應(yīng)用程序可以從隊(duì)列中檢索消息，而無(wú)需知道誰(shuí)發(fā)送了這些消息。消息隊(duì)列是一種重要的中間件，它可以幫助應(yīng)用程序之間進(jìn)行異步、可靠、可擴(kuò)展的通信。常見(jiàn)的消息隊(duì)列中間件有ActiveMQ，RabbitMQ，Kafka......今天我們就來(lái)介紹RabbitMQ。

2、什么是RabbitMQ

RabbitMQ 是一個(gè)開(kāi)源的消息隊(duì)列服務(wù)器，它實(shí)現(xiàn)了 AMQP (高級(jí)消息隊(duì)列協(xié)議) 標(biāo)準(zhǔn)。AMQP 是一種應(yīng)用層協(xié)議，為面向消息的中間件設(shè)計(jì)，基于此協(xié)議的客戶端與消息中間件可傳遞消息，并不受產(chǎn)品、開(kāi)發(fā)語(yǔ)言等條件的限制。

AMQP ：Advanced Message Queue，高級(jí)消息隊(duì)列協(xié)議。它是應(yīng)用層協(xié)議的一個(gè)開(kāi)放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計(jì)，基于此協(xié)議的客戶端與消息中間件可傳遞消息，并不受產(chǎn)品、開(kāi)發(fā)語(yǔ)言等條件的限制。

RabbitMQ 的主要特點(diǎn)包括：

高性能：RabbitMQ 能夠處理大量的消息，并提供低延遲的性能。
可靠性：RabbitMQ 提供持久化消息存儲(chǔ)，確保消息不會(huì)丟失。
可擴(kuò)展性：RabbitMQ 可以輕松擴(kuò)展以滿足不斷增長(zhǎng)的需求。
靈活性：RabbitMQ 支持多種編程語(yǔ)言和客戶端，并提供豐富的功能和配置選項(xiàng)。

RabbitMQ 的常見(jiàn)應(yīng)用場(chǎng)景包括：

分布式系統(tǒng)：RabbitMQ 可以用于在分布式系統(tǒng)中進(jìn)行異步通信。
異步處理：RabbitMQ 可以用于異步處理任務(wù)，提高系統(tǒng)的性能和效率。
消息隊(duì)列：RabbitMQ 可以用于實(shí)現(xiàn)消息隊(duì)列，例如任務(wù)隊(duì)列、發(fā)布/訂閱隊(duì)列等。
消息通知：RabbitMQ 可以用于發(fā)送消息通知，例如電子郵件或短信。

3、安裝RabbitMQ

由于RabbitMQ是一個(gè)由 Erlang 語(yǔ)言開(kāi)發(fā)的 AMQP 的開(kāi)源實(shí)現(xiàn)。所以在安裝RabbitMQ前需要先安裝Erlang環(huán)境。

Erlang下載地址：Downloads - Erlang/OTP

RabbitMQ下載地址：Installing RabbitMQ | RabbitMQ

先安裝Erlang，在安裝RabbitMQ。安裝工程相對(duì)簡(jiǎn)單，無(wú)腦下一步即可。

安裝完RabbitMQ后，打開(kāi)cmd窗口，進(jìn)入RabbitMQ的安裝目錄的sbin下，我的目錄是：

D:\RabbitMQ Server\rabbitmq_server-3.13.0\sbin

然后輸入以下命令安裝一下插件：

rabbitmq-plugins enable rabbitmq_management

提示以下這個(gè)就是安裝成功。

驗(yàn)證RabbitMQ是否安裝成功，輸入以下命令：

rabbitmqctl status

這時(shí)候，直接訪問(wèn)http://127.0.0.1:15672就可以看到RabbitMQ的管理頁(yè)面了，RabbitMQ默認(rèn)端口為15672，默認(rèn)的管理頁(yè)面賬號(hào)密碼均為guest。

登錄后，就可以看到一個(gè)初始的管理界面：

4、Springboot集成RabbitMQ

4.1、添加依賴

<?xml version="1.0" encoding="UTF-8"?>
<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-parent</artifactId>
       <version>3.2.3</version>
       <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>springboot-rabbitmq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-rabbitmq</name>
    <description>springboot-rabbitmq</description>
    <properties>
       <java.version>17</java.version>
    </properties>
    <dependencies>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter</artifactId>
       </dependency>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-devtools</artifactId>
          <scope>runtime</scope>
          <optional>true</optional>
       </dependency>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-configuration-processor</artifactId>
          <optional>true</optional>
       </dependency>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-test</artifactId>
          <scope>test</scope>
       </dependency>
       <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>cn.hutool</groupId>
          <artifactId>hutool-all</artifactId>
          <version>5.8.24</version>
       </dependency>
    </dependencies>
    <build>
       <plugins>
          <plugin>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-maven-plugin</artifactId>
          </plugin>
       </plugins>
    </build>
</project>

4.2、添加配置

# rabbitmq連接配置信息
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
# 確保消息在未被隊(duì)列接收時(shí)返回
spring.rabbitmq.publisher-returns=true
# 發(fā)布消息成功到交換器后會(huì)觸發(fā)回調(diào)方法
spring.rabbitmq.publisher-confirm-type=correlated

4.3、添加controller，作為生產(chǎn)者

新建controller，用于發(fā)送消息。

package com.example.springbootrabbitmq.controller;
import com.example.springbootrabbitmq.config.MqProducerCallBack;
import jakarta.annotation.Resource;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("push/message")
public class PushMessageController {
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private MqProducerCallBack mqProducerCallBack;
    @GetMapping("test")
    public String sendMessage() {
        // correlationData：對(duì)象內(nèi)部只有一個(gè) id 屬性，用來(lái)表示當(dāng)前消息的唯一性。
        CorrelationData correlationData = new CorrelationData("id_" + System.currentTimeMillis());
        // 消息確認(rèn)和返回回調(diào)
        rabbitTemplate.setConfirmCallback(mqProducerCallBack);
        rabbitTemplate.setReturnsCallback(mqProducerCallBack);
        // 消息發(fā)送
        rabbitTemplate.convertAndSend("my-queue", "hello world", message -> {
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return message;
        }, correlationData);
        return "publisher success...";
    }
}

4.4、設(shè)置生產(chǎn)者消息確認(rèn)CallBack

package com.example.springbootrabbitmq.config;
import cn.hutool.json.JSONUtil;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
@Component
public class MqProducerCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
    /**
     * correlationData：對(duì)象內(nèi)部只有一個(gè) id 屬性，用來(lái)表示當(dāng)前消息的唯一性。
     * ack：消息投遞到broker 的狀態(tài)，true成功，false失敗。
     * cause：投遞失敗的原因。
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack) {
            System.err.println("消息ID=" + correlationData.getId() + "投遞失敗，失敗原因：" + cause);
        } else {
            System.out.println("消息投遞收到確認(rèn)，correlationData=" + correlationData.getId());
        }
    }
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        System.out.println("返回消息結(jié)果：" + JSONUtil.toJsonStr(returnedMessage));
    }
}

4.5、添加Consumer，作為消費(fèi)者

package com.example.springbootrabbitmq.consumer;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Component
public class PushMessageConsumer {
    /**
     * basicAck：表示成功確認(rèn)，使用此回執(zhí)方法后，消息會(huì)被rabbitmq broker 刪除。
     * void basicAck(long deliveryTag, boolean multiple)
     * deliveryTag：表示消息投遞序號(hào)，每次消費(fèi)消息或者消息重新投遞后，deliveryTag都會(huì)增加。手動(dòng)消息確認(rèn)模式下，我們可以對(duì)指定deliveryTag的消息進(jìn)行ack、nack、reject等操作。
     * multiple：是否批量確認(rèn)，值為 true 則會(huì)一次性 ack所有小于當(dāng)前消息 deliveryTag 的消息。
     * */
    @RabbitListener(queuesToDeclare = @Queue(value = "my-queue"))
    @RabbitHandler
    public void consume(String msg, Channel channel, Message message) throws IOException {
        try {
            System.out.println("消費(fèi)者收到消息：" + msg);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("deliveryTag：" + message.getMessageProperties().getDeliveryTag());
            System.out.println("redelivered：" + message.getMessageProperties().getRedelivered());
        } catch (Exception e) {
            if (message.getMessageProperties().getRedelivered()) {
                System.err.println("消息已重復(fù)處理失敗,拒絕再次接收！");
                /**
                 * 拒絕消息，requeue=false 表示不再重新入隊(duì)，如果配置了死信隊(duì)列則進(jìn)入死信隊(duì)列
                 * basicReject：拒絕消息，與basicNack區(qū)別在于不能進(jìn)行批量操作，其他用法很相似。
                 * deliveryTag：表示消息投遞序號(hào)。
                 * requeue：值為 true 消息將重新入隊(duì)列。
                 */
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                System.out.println("消息即將再次返回隊(duì)列處理！");
                /**
                 * requeue為是否重新回到隊(duì)列，true重新入隊(duì)
                 * deliveryTag：表示消息投遞序號(hào)。
                 * multiple：是否批量確認(rèn)。
                 * requeue：值為 true 消息將重新入隊(duì)列。
                 */
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
        }
    }
}