RabbitMQ 核心概念

在分布式系统开发中，消息队列是实现系统解耦、削峰填谷、异步通信的核心组件。RabbitMQ作为一款基于AMQP协议的开源消息中间件，凭借其轻量、可靠、灵活的特性，被广泛应用于各类企业级项目。而Spring Boot通过自动配置，极大简化了RabbitMQ的整合流程，让开发者能够快速上手。

1. RabbitMQ核心概念

在动手实操前，先明确RabbitMQ的核心组件和工作流程。

1.1 核心角色

生产者（Producer）：发送消息的应用程序；
消费者（Consumer）：接收并处理消息的应用程序；
交换机（Exchange）：接收生产者发送的消息，根据路由规则将消息转发到对应队列；
队列（Queue）：存储消息的容器，消息最终到达队列后被消费者监听消费；
绑定（Binding）：建立交换机与队列之间的关联，同时指定路由规则；
虚拟主机（VHost）：相当于RabbitMQ的”命名空间”，隔离不同项目的资源（队列、交换机等）。

1.2 交换机类型（核心重点）

RabbitMQ的路由能力依赖于交换机类型，不同类型对应不同的路由规则，实际开发中需根据场景选择：

Direct（直连交换机）：最常用，根据路由键（Routing Key）精准匹配队列，只有路由键完全一致时才会转发消息（适合精准路由场景，如订单支付结果通知）；
Fanout（广播交换机）：不依赖路由键，将消息广播到所有绑定的队列（适合全局通知场景，如系统公告推送）；
Topic（主题交换机）：支持模糊匹配路由键，使用通配符 *（匹配一个单词）和#（匹配多个单词，含0个），灵活性最高（适合按规则路由场景，如用户行为日志分类）；
Headers（头匹配交换机）：根据消息头信息匹配队列，不依赖路由键，较少使用。

1.3 核心工作流程

生产者 → 交换机（根据绑定规则）→ 队列 → 消费者（监听消费）

2. 环境安装

2.1 Windows安装

2.1.1 安装Erlang

RabbitMQ基于Erlang语言开发，因此安装前需先配置Erlang环境。

下载Erlang安装包：访问Erlang官网，选择与RabbitMQ兼容的版本（版本兼容对照表：RabbitMQ官方兼容表）；
双击安装，选择安装路径，一路下一步即可；
配置环境变量：新建系统变量 ERLANG_HOME，值为Erlang安装路径（如 D:\Program Files\Erlang OTP），然后在Path中添加 %ERLANG_HOME%\bin；
验证：打开cmd，输入 erl -version，若显示版本信息则安装成功。

2.1.2 安装RabbitMQ

下载RabbitMQ安装包：访问RabbitMQ官网，选择Windows版本；
双击安装，选择安装路径，一路下一步（默认会自动配置环境变量）；
安装管理插件（可视化控制台）：打开cmd，进入RabbitMQ安装目录的sbin文件夹（如 D:\Program Files\RabbitMQ Server\rabbitmq_server-4.2.0\sbin），输入命令 rabbitmq-plugins enable rabbitmq_management，等待插件安装完成；
启动RabbitMQ服务：cmd（RabbitMQ Command Prompt）输入 rabbitmq-server start（启动），或通过Windows服务面板找到”RabbitMQ”服务，设置为自动启动并启动。

2.1.3 安装验证

访问可视化控制台：打开浏览器，输入 http://localhost:15672（管理端默认端口15672，服务端默认端口5672）；
登录：默认账号密码 guest/guest（仅本地访问可用），登录成功后即可看到RabbitMQ的控制台界面；
命令行验证：cmd（RabbitMQ Command Prompt）输入 rabbitmqctl status，若显示服务状态信息则服务正常。

3. Spring Boot 整合 RabbitMQ 核心实操

3.1 基础准备

3.1.1 创建 Spring Boot 项目

通过 Spring Initializr 创建项目，选择核心依赖：

Spring Web（用于编写测试接口）；
Spring AMQP（RabbitMQ 整合核心依赖）；
Lombok（简化实体类编写，可选）。

3.1.2 配置文件（application.yml）

在resources目录下创建application.yml，完善 RabbitMQ 连接及可靠性配置：

spring:
  rabbitmq:
    host: localhost # 服务端地址（生产环境替换为实际IP）
    port: 5672 # 默认端口
    username: admin # 账号（Docker安装为admin，原生RabbitMQ默认guest）
    password: admin # 密码（与账号对应）
    virtual-host: / # 虚拟主机（资源隔离，默认/）
    connection-timeout: 15000 # 连接超时时间（毫秒）
    # 生产者可靠性配置
    publisher-confirm-type: correlated # 异步确认模式，触发ConfirmCallback
    publisher-returns: true # 开启消息返回机制，路由失败触发ReturnsCallback
    # 消费者可靠性配置
    listener:
      simple:
        acknowledge-mode: manual # 手动ACK，确保业务处理完成后确认消息
        prefetch: 10 # 限流：每次最多拉取10条消息，避免消费者过载
        concurrency: 1 # 消费者最小线程数
        max-concurrency: 5 # 消费者最大线程数

3.2 Topic 交换机（模糊路由）实战

Topic 交换机是最灵活的交换机类型，支持通配符模糊路由：

*：匹配单个单词（如business.*匹配business.user，不匹配business.user.login）；
#：匹配多个单词（如business.#匹配business.user/business.user.login）。

结合死信队列实现消息可靠性保障，完整流程如下：

3.2.1 核心配置类

3.2.1.1 队列 / 交换机 / 绑定关系配置

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMQ核心配置：声明队列、交换机、绑定关系（含死信队列）
 */
@Configuration
public class RabbitMQConfig {

    // ========== 业务队列相关配置 ==========
    public static final String BUSINESS_QUEUE_NAME = "business_queue";
    public static final String BUSINESS_EXCHANGE_NAME = "business_exchange";
    public static final String BUSINESS_ROUTING_KEY = "business.#"; // 模糊路由键

    // ========== 死信队列相关配置 ==========
    public static final String DLX_QUEUE_NAME = "dlx_queue";
    public static final String DLX_EXCHANGE_NAME = "dlx_exchange";
    public static final String DLX_ROUTING_KEY = "dlx.routing.key";

    /**
     * 业务队列（持久化+绑定死信）
     * 死信触发场景：消费失败NACK、消息TTL过期、队列满溢出
     */
    @Bean
    public Queue businessQueue() {
        return QueueBuilder.durable(BUSINESS_QUEUE_NAME)
                .deadLetterExchange(DLX_EXCHANGE_NAME) // 绑定死信交换机
                .deadLetterRoutingKey(DLX_ROUTING_KEY) // 死信路由键
                // .ttl(60000) // 可选：消息过期时间（毫秒）
                // .maxLength(1000) // 可选：队列最大长度
                .build();
    }

    /**
     * Topic交换机（持久化）
     */
    @Bean
    public TopicExchange businessExchange() {
        return ExchangeBuilder.topicExchange(BUSINESS_EXCHANGE_NAME)
                .durable(true) // 持久化，服务重启不丢失
                .build();
    }

    /**
     * 业务队列与Topic交换机绑定（模糊路由）
     */
    @Bean
    public Binding businessBinding(Queue businessQueue, TopicExchange businessExchange) {
        return BindingBuilder.bind(businessQueue)
                .to(businessExchange)
                .with(BUSINESS_ROUTING_KEY);
    }

    /**
     * 死信队列（存储消费失败/过期消息）
     */
    @Bean
    public Queue dlxQueue() {
        return QueueBuilder.durable(DLX_QUEUE_NAME).build();
    }

    /**
     * 死信交换机
     */
    @Bean
    public TopicExchange dlxExchange() {
        return ExchangeBuilder.topicExchange(DLX_EXCHANGE_NAME)
                .durable(true)
                .build();
    }

    /**
     * 死信队列与死信交换机绑定
     */
    @Bean
    public Binding dlxBinding(Queue dlxQueue, TopicExchange dlxExchange) {
        return BindingBuilder.bind(dlxQueue)
                .to(dlxExchange)
                .with(DLX_ROUTING_KEY);
    }
}

3.2.1.2 生产者确认回调配置

import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

/**
 * 生产者确认机制全局配置（@PostConstruct初始化）
 */
@Component
public class ProducerConfirmConfig {
    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * Spring容器初始化Bean后执行（依赖注入完成）
     * 初始化ConfirmCallback（交换机确认）和ReturnsCallback（路由确认）
     */
    @PostConstruct
    public void initProducerConfirm() {
        // 1. ConfirmCallback：确认消息是否到达交换机
        rabbitTemplate.setConfirmCallback((CorrelationData correlationData, boolean ack, String cause) -> {
            String msgId = correlationData != null ? correlationData.getId() : "未知ID";
            if (ack) {
                System.out.println("[生产者确认] 消息到达交换机，消息ID：" + msgId);
            } else {
                System.err.println("[生产者确认] 消息未到达交换机，消息ID：" + msgId + "，原因：" + cause);
                // 补偿逻辑：重试发送、持久化到失败表、告警通知等
            }
        });

        // 2. ReturnsCallback：确认消息是否路由到队列（仅到达交换机但路由失败时触发）
        rabbitTemplate.setReturnsCallback(returned -> {
            String msgContent = new String(returned.getMessage().getBody());
            System.err.println("[生产者返回] 消息路由失败");
            System.err.println("消息内容：" + msgContent);
            System.err.println("错误码：" + returned.getReplyCode() + "，错误信息：" + returned.getReplyText());
            System.err.println("交换机：" + returned.getExchange() + "，路由键：" + returned.getRoutingKey());
            // 补偿逻辑：重新发送、记录日志、人工介入等
        });
    }
}

3.2.1.3 消息 JSON 序列化配置

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 消息转换器配置：替代默认JDK序列化，使用JSON序列化（可读性高、体积小）
 */
@Configuration
public class RabbitMessageConverterConfig {

    @Bean
    public MessageConverter messageConverter() {
        // Spring AMQP 4.0+ 推荐使用Jackson2JsonMessageConverter（替代弃用的JacksonJsonMessageConverter）
        // 注意：发送的对象需符合JavaBean规范（无参构造、getter/setter）
        return new Jackson2JsonMessageConverter();
    }
}

3.2.2 生产者（发送消息）

import com.hnu.springbootlearning.mq.config.RabbitMQConfig;
import jakarta.annotation.Resource;
import lombok.Data;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * 通用生产者：支持字符串/对象消息发送
 */
@Component
public class RabbitMQProducer {
    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送字符串消息
     * @param message 消息内容
     */
    public void sendSimpleMessage(String message) {
        // CorrelationData：消息唯一标识，用于关联发送与确认结果
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(
                RabbitMQConfig.BUSINESS_EXCHANGE_NAME,
                RabbitMQConfig.BUSINESS_ROUTING_KEY,
                message,
                correlationData
        );
    }

    /**
     * 重载：发送字符串消息（指定路由键，测试模糊路由）
     * @param message 消息内容
     * @param routingKey 路由键
     */
    public void sendSimpleMessage(String message, String routingKey) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(
                RabbitMQConfig.BUSINESS_EXCHANGE_NAME,
                routingKey,
                message,
                correlationData
        );
    }

    /**
     * 发送对象消息（自动序列化为JSON）
     * @param user 待发送对象（需符合JavaBean规范）
     */
    public void sendObjectMessage(User user) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(
                RabbitMQConfig.BUSINESS_EXCHANGE_NAME,
                RabbitMQConfig.BUSINESS_ROUTING_KEY,
                user,
                correlationData
        );
    }

    /**
     * 示例：用户实体类（Lombok简化getter/setter）
     */
    @Data
    public static class User {
        private Long id;
        private String name;
        // 可扩展其他字段：age、email等
    }
}

3.2.3 消费者（监听消息，手动 ACK）

import com.hnu.springbootlearning.mq.config.RabbitMQConfig;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * 消费者：监听业务队列+死信队列，手动ACK确保消息可靠消费
 */
@Component
public class RabbitMQConsumer {

    /**
     * 监听业务队列（兼容字符串/对象消息）
     * @param message 原始消息对象（万能接收，避免类型不匹配）
     * @param channel 信道对象（用于手动ACK/NACK）
     */
    @RabbitListener(queues = RabbitMQConfig.BUSINESS_QUEUE_NAME)
    public void consumeBusinessQueue(Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 解析消息内容（JSON对象可在此反序列化为JavaBean）
            String content = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.println("[业务队列] 接收到消息：" + content);

            // 模拟业务逻辑（实际开发替换为真实业务：如更新订单、发送通知等）
            // 测试死信触发：包含"死信测试"的消息手动抛异常
            if (content.contains("死信测试")) {
                throw new RuntimeException("模拟消费失败，触发死信队列");
            }

            // 手动ACK：确认消息消费成功，队列删除消息
            // multiple=false：仅确认当前消息；true：确认当前及之前所有未确认消息
            channel.basicAck(deliveryTag, false);
            System.out.println("[业务队列] 消息ACK成功，deliveryTag：" + deliveryTag);
        } catch (Exception e) {
            System.err.println("[业务队列] 消费失败：" + e.getMessage() + "，deliveryTag：" + deliveryTag);
            // 手动NACK：消费失败，拒绝消息并转发到死信队列（requeue=false）
            channel.basicNack(deliveryTag, false, false);
        }
    }

    /**
     * 监听死信队列（处理消费失败/过期消息）
     * 死信消息建议：记录日志、持久化、人工干预，一般不重新入队
     */
    @RabbitListener(queues = RabbitMQConfig.DLX_QUEUE_NAME)
    public void consumeDlxQueue(Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            String content = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.println("[死信队列] 接收到消息：" + content);

            // 死信处理逻辑：如持久化到数据库、发送告警、定时重试等
            // handleDlxMessage(content);

            channel.basicAck(deliveryTag, false);
            System.out.println("[死信队列] 消息ACK成功，deliveryTag：" + deliveryTag);
        } catch (Exception e) {
            System.err.println("[死信队列] 消费失败：" + e.getMessage() + "，deliveryTag：" + deliveryTag);
            // 死信消息再次失败：直接ACK删除，避免无限循环
            channel.basicAck(deliveryTag, false);
        }
    }

    // 示例：死信消息处理方法（实际开发抽离到Service层）
    private void handleDlxMessage(String content) {
        // 1. 记录死信日志到数据库
        // 2. 发送钉钉/邮件告警
        // 3. 人工介入处理
    }
}

3.2.4 测试用例（完整场景验证）

import com.hnu.springbootlearning.mq.demo.producer.RabbitMQProducer;
import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.concurrent.TimeUnit;

/**
 * RabbitMQ集成测试：按场景拆分独立测试方法，便于定位问题
 * 注解说明：
 * - @SpringBootTest：启动Spring容器，加载所有Bean
 * - 每个@Test方法独立执行，互不干扰
 */
@SpringBootTest
public class RabbitMQTest {

    @Resource
    private RabbitMQProducer producer;

    /**
     * 测试1：基础字符串消息发送（固定路由键）
     * 验证：消息正常发送→到达交换机→业务队列消费→ACK
     */
    @Test
    public void testBasicStringMessage() throws InterruptedException {
        // 发送固定路由键的字符串消息
        producer.sendSimpleMessage("Hello RabbitMQ!");
        // 延迟等待异步回调和消费完成（每个测试独立延迟，避免依赖）
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 测试2：Topic交换机模糊路由
     * 验证：business.# 能匹配 business.user.login 等多级路由键
     */
    @Test
    public void testTopicFuzzyRouting() throws InterruptedException {
        // 发送多级路由键消息，验证模糊匹配
        producer.sendSimpleMessage("测试Topic模糊路由", "business.user.login");
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 测试3：消费失败触发死信队列
     * 验证：业务队列消费失败→NACK→消息转发到死信队列→死信队列消费
     */
    @Test
    public void testDlxQueueTrigger() throws InterruptedException {
        // 发送含"死信测试"的消息，消费者会模拟异常触发死信
        producer.sendSimpleMessage("死信测试：消费失败触发", "business.dlx.test");
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 测试4：对象消息JSON序列化
     * 验证：Java对象自动序列化为JSON→消息传输→消费者正常解析
     */
    @Test
    public void testObjectMessageJsonSerialize() throws InterruptedException {
        // 构建用户对象并发送
        RabbitMQProducer.User user = new RabbitMQProducer.User();
        user.setId(1L);
        user.setName("张三");
        producer.sendObjectMessage(user);
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 测试5：路由失败触发ReturnsCallback
     * 验证：消息到达交换机但路由键不匹配→触发Returns回调
     */
    @Test
    public void testRoutingFailReturnCallback() throws InterruptedException {
        // 发送不存在的路由键，触发路由失败
        producer.sendSimpleMessage("测试路由失败", "error.routing.key");
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 可选：全场景批量测试（保留原有逻辑，便于一次性验证所有场景）
     */
    @Test
    public void testAllScenarios() throws InterruptedException {
        testBasicStringMessage();
        testTopicFuzzyRouting();
        testDlxQueueTrigger();
        testObjectMessageJsonSerialize();
        testRoutingFailReturnCallback();
        TimeUnit.SECONDS.sleep(1);
    }
}

3.2.5 验证结果

3.2.5.1 正常场景输出

[业务队列] 接收到消息："Hello RabbitMQ!"
[业务队列] 消息ACK成功，deliveryTag：8
[生产者确认] 消息到达交换机，消息ID：93e9e82d-5dc7-4208-8d27-d86a9da88c20

[业务队列] 接收到消息："测试Topic模糊路由"
[业务队列] 消息ACK成功，deliveryTag：7
[生产者确认] 消息到达交换机，消息ID：a77747d6-ad44-4434-bd3e-ad038ceb1289

[业务队列] 接收到消息："死信测试：消费失败触发"
[业务队列] 消费失败：模拟消费失败，触发死信队列，deliveryTag：6
[生产者确认] 消息到达交换机，消息ID：ec8675de-4c4b-472f-8b9d-167cd0750f98
[死信队列] 接收到消息："死信测试：消费失败触发"
[死信队列] 消息ACK成功，deliveryTag：2

[生产者确认] 消息到达交换机，消息ID：80ada516-5b74-4f4c-85dc-a811a652197c
[业务队列] 接收到消息：{"id":1,"name":"张三"}
[业务队列] 消息ACK成功，deliveryTag：1

3.2.5.2 路由失败场景输出

[生产者返回] 消息路由失败
消息内容："测试路由失败"
错误码：312，错误信息：NO_ROUTE
交换机：business_exchange，路由键：error.routing.key
[生产者确认] 消息到达交换机，消息ID：ca047796-0231-44d3-ac6d-a024c96ffdc1

3.2.6 常见问题与注意事项

手动 ACK 遗漏：忘记调用basicAck会导致消息停留在Unacked状态，队列堆积，需确保业务完成后显式确认；
死信队列堆积：定期监控死信队列长度，失败消息建议持久化到数据库并触发人工干预，避免无限循环；
JSON 序列化失败：发送的对象必须符合 JavaBean 规范（无参构造、getter/setter），否则反序列化报错；
生产者确认不触发：必须在配置文件开启publisher-confirm-type: correlated和publisher-returns: true；
限流配置合理值：prefetch根据消费者处理能力调整（如 CPU 密集型设 5，IO 密集型设 20），避免 OOM；
虚拟主机权限：生产环境需给用户分配指定虚拟主机的权限，否则连接失败；
消息幂等性：消费者需实现幂等逻辑（如基于消息 ID 去重），避免重复消费导致数据不一致。

3.2.7 生产环境扩展建议

消息追踪：集成 RabbitMQ Trace 插件或自定义日志，全链路追踪消息生命周期；
监控告警：监控队列长度、消费速率、死信数量，超过阈值触发告警；
重试机制：使用 Spring Retry 实现生产者重试，避免网络抖动导致消息丢失；
批量消费：高并发场景使用batchListener批量拉取消息，提升消费效率；
动态配置：通过配置中心（Nacos/Apollo）动态调整消费者线程数、限流阈值等参数。