一文掌握Springboot集成RabbitMQ的方法_Java

1、前言

消息队列（message queue，简称 mq）是一种异步的消息传递中间件，它解耦了应用程序之间的通信。应用程序可以将消息发送到队列，而无需知道谁会接收这些消息。接收应用程序可以从队列中检索消息，而无需知道谁发送了这些消息。消息队列是一种重要的中间件，它可以帮助应用程序之间进行异步、可靠、可扩展的通信。常见的消息队列中间件有activemq，rabbitmq，kafka......今天我们就来介绍rabbitmq。

2、什么是rabbitmq

rabbitmq 是一个开源的消息队列服务器，它实现了 amqp (高级消息队列协议) 标准。amqp 是一种应用层协议，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。

amqp ：advanced message queue，高级消息队列协议。它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。

rabbitmq 的主要特点包括：

高性能：rabbitmq 能够处理大量的消息，并提供低延迟的性能。
可靠性：rabbitmq 提供持久化消息存储，确保消息不会丢失。
可扩展性：rabbitmq 可以轻松扩展以满足不断增长的需求。
灵活性：rabbitmq 支持多种编程语言和客户端，并提供丰富的功能和配置选项。

rabbitmq 的常见应用场景包括：

分布式系统：rabbitmq 可以用于在分布式系统中进行异步通信。
异步处理：rabbitmq 可以用于异步处理任务，提高系统的性能和效率。
消息队列：rabbitmq 可以用于实现消息队列，例如任务队列、发布/订阅队列等。
消息通知：rabbitmq 可以用于发送消息通知，例如电子邮件或短信。

3、安装rabbitmq

由于rabbitmq是一个由 erlang 语言开发的 amqp 的开源实现。所以在安装rabbitmq前需要先安装erlang环境。

erlang下载地址：downloads - erlang/otp

rabbitmq下载地址：installing rabbitmq | rabbitmq

先安装erlang，在安装rabbitmq。安装工程相对简单，无脑下一步即可。

安装完rabbitmq后，打开cmd窗口，进入rabbitmq的安装目录的sbin下，我的目录是：

d:\rabbitmq server\rabbitmq_server-3.13.0\sbin

然后输入以下命令安装一下插件：

rabbitmq-plugins enable rabbitmq_management

提示以下这个就是安装成功。

验证rabbitmq是否安装成功，输入以下命令：

rabbitmqctl status

这时候，直接访问http://127.0.0.1:15672就可以看到rabbitmq的管理页面了，rabbitmq默认端口为15672，默认的管理页面账号密码均为guest。

登录后，就可以看到一个初始的管理界面：

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
# 确保消息在未被队列接收时返回
spring.rabbitmq.publisher-returns=true
# 发布消息成功到交换器后会触发回调方法
spring.rabbitmq.publisher-confirm-type=correlated

4.3、添加controller，作为生产者

新建controller，用于发送消息。

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：对象内部只有一个 id 属性，用来表示当前消息的唯一性。
        correlationdata correlationdata = new correlationdata("id_" + system.currenttimemillis());
        // 消息确认和返回回调
        rabbittemplate.setconfirmcallback(mqproducercallback);
        rabbittemplate.setreturnscallback(mqproducercallback);
        // 消息发送
        rabbittemplate.convertandsend("my-queue", "hello world", message -> {
            message.getmessageproperties().setdeliverymode(messagedeliverymode.persistent);
            return message;
        }, correlationdata);
        return "publisher success...";
    }
}

4.4、设置生产者消息确认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：对象内部只有一个 id 属性，用来表示当前消息的唯一性。
     * ack：消息投递到broker 的状态，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("消息投递收到确认，correlationdata=" + correlationdata.getid());
        }
    }
    @override
    public void returnedmessage(returnedmessage returnedmessage) {
        system.out.println("返回消息结果：" + jsonutil.tojsonstr(returnedmessage));
    }
}

4.5、添加consumer，作为消费者

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：表示成功确认，使用此回执方法后，消息会被rabbitmq broker 删除。
     * void basicack(long deliverytag, boolean multiple)
     * deliverytag：表示消息投递序号，每次消费消息或者消息重新投递后，deliverytag都会增加。手动消息确认模式下，我们可以对指定deliverytag的消息进行ack、nack、reject等操作。
     * multiple：是否批量确认，值为 true 则会一次性 ack所有小于当前消息 deliverytag 的消息。
     * */
    @rabbitlistener(queuestodeclare = @queue(value = "my-queue"))
    @rabbithandler
    public void consume(string msg, channel channel, message message) throws ioexception {
        try {
            system.out.println("消费者收到消息：" + 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("消息已重复处理失败,拒绝再次接收！");
                /**
                 * 拒绝消息，requeue=false 表示不再重新入队，如果配置了死信队列则进入死信队列
                 * basicreject：拒绝消息，与basicnack区别在于不能进行批量操作，其他用法很相似。
                 * deliverytag：表示消息投递序号。
                 * requeue：值为 true 消息将重新入队列。
                 */
                channel.basicreject(message.getmessageproperties().getdeliverytag(), false);
            } else {
                system.out.println("消息即将再次返回队列处理！");
                /**
                 * requeue为是否重新回到队列，true重新入队
                 * deliverytag：表示消息投递序号。
                 * multiple：是否批量确认。
                 * requeue：值为 true 消息将重新入队列。
                 */
                channel.basicnack(message.getmessageproperties().getdeliverytag(), false, true);
            }
        }
    }
}