SpringBoot集成RabbitMQ企业级实战教程

本文将从零开始，基于 IntelliJ IDEA 2023 Ultimate 版，搭建企业级的 SpringBoot-RabbitMQ 项目，涵盖 RabbitMQ 核心用法和企业级特性，所有代码可直接复制使用，适合新手快速上手并落地到生产环境。

1. 开发环境准备

1.1 环境清单（企业级标准配置）

工具/框架	版本号	备注
IntelliJ IDEA	2023 Ultimate	项目管理和代码开发
JDK	17	SpringBoot 3.x 最低要求
SpringBoot	3.5.10	核心开发框架
Maven	3.9.4	项目构建和依赖管理
RabbitMQ	3.13.2	消息中间件
操作系统	Windows/MacOS	无特殊限制

1.2 RabbitMQ 安装与验证（Docker 快速部署）

企业级环境推荐使用 Docker 部署 RabbitMQ（含管理插件），命令如下：

# 拉取 RabbitMQ 3.13.2 镜像（带管理插件）
docker pull rabbitmq:3.13.2-management

# 启动容器（映射端口、配置用户名密码，与你提供的信息对齐）
docker run -d --name rabbitmq-server \
  -p 2765:5672 \  # AMQP 协议端口（映射为你指定的2765）
  -p 15672:15672 \ # 管理界面端口
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=Admin@1029384756 \
  rabbitmq:3.13.2-management

验证安装：

1. 访问 RabbitMQ 管理界面：http://localhost:15672（你的地址+管理端口）
2. 输入用户名 admin、密码 Admin@1029384756，能登录则说明部署成功。

2. 项目初始化（IDEA 操作）

2.1 新建 SpringBoot 项目

2.2 验证 Maven 配置

项目根目录的 pom.xml 需包含以下核心依赖（自动生成，确认版本即可）：

<?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.5.10</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>
	<groupId>cc.hep</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>
		<!-- Spring Web 依赖 -->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<!-- RabbitMQ 核心依赖 -->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>
		<!-- 测试依赖 -->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<!-- RabbitMQ 测试依赖 -->
		<dependency>
			<groupId>org.springframework.amqp</groupId>
			<artifactId>spring-rabbit-test</artifactId>
			<scope>test</scope>
		</dependency>
		<!-- Jackson（Spring Boot 默认包含，这里显式声明确保版本） -->
		<dependency>
			<groupId>com.fasterxml.jackson.core</groupId>
			<artifactId>jackson-databind</artifactId>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
				<configuration>
					<excludes>
						<exclude>
							<groupId>org.projectlombok</groupId>
							<artifactId>lombok</artifactId>
						</exclude>
					</excludes>
				</configuration>
			</plugin>
		</plugins>
	</build>
</project>

3. RabbitMQ 核心配置（企业级规范）

3.1 应用配置文件（application.yml）

在 src/main/resources 下创建 application.yml，配置 RabbitMQ 连接信息和企业级参数（连接池、消费策略等）：

spring:
  # RabbitMQ 核心配置
  rabbitmq:
    host: localhost  # 你的RabbitMQ地址
    port: 2765           # 你的RabbitMQ端口
    username: admin      # 用户名
    password: Admin@1029384756  # 密码
    virtual-host: /      # 虚拟主机（默认/）

    # 增强连接稳定性配置
    connection-timeout: 30000  # 增加连接超时到30秒
    requested-heartbeat: 60    # 心跳间隔60秒（防止超时断开）

    # 连接池配置
    cache:
      mode: CONNECTION
      connection:
        size: 5        # 减少连接池大小，避免过多连接
        mode: CONNECTION
        channel:
          checkout-timeout: 30000  # 通道获取超时

    # 自动重连配置
    template:
      retry:
        enabled: true    # 开启发送重试
        max-attempts: 3  # 最大重试次数
        initial-interval: 1000  # 初始重试间隔1秒

    # 消费端配置（企业级核心：手动ACK，避免消息丢失）
    listener:
      simple:
        acknowledge-mode: manual  # 手动确认消息
        concurrency: 3            # 减少消费线程数（从5减到3）
        max-concurrency: 5        # 最大消费线程数
        prefetch: 5               # 减少预取数量（从10减到5）
        retry:
          enabled: true           # 开启消费重试
          max-attempts: 3         # 最大重试次数
          initial-interval: 2000  # 首次重试间隔（2秒）
        # 自动恢复配置
        auto-startup: true        # 自动启动消费者
        missing-queues-fatal: false  # 队列不存在时不致命
        recovery-interval: 5000   # 恢复间隔5秒

# 自定义RabbitMQ常量
rabbitmq:
  exchange:
    direct: "hep.direct.exchange"
    topic: "hep.topic.exchange"
  queue:
    normal: "hep.normal.queue"
    dead: "hep.dead.queue"
  routing-key:
    normal: "hep.normal.routing.key"
    dead: "hep.dead.routing.key"

3.2 RabbitMQ 核心配置类（企业级声明）

在 cc.hep.config 包下创建 RabbitMQConfig，用于声明交换机、队列、绑定关系（企业级规范：代码声明而非手动创建）：

package cc.hep.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMQ 核心配置类（企业级规范：声明交换机、队列、绑定关系）
 */
@Configuration
public class RabbitMQConfig {

    // 从配置文件读取常量
    @Value("${rabbitmq.exchange.direct}")
    private String directExchange;

    @Value("${rabbitmq.exchange.topic}")
    private String topicExchange;

    @Value("${rabbitmq.queue.normal}")
    private String normalQueue;

    @Value("${rabbitmq.queue.dead}")
    private String deadQueue;

    @Value("${rabbitmq.routing-key.normal}")
    private String normalRoutingKey;

    @Value("${rabbitmq.routing-key.dead}")
    private String deadRoutingKey;

    /**
     * 1. 声明直连交换机（企业级常用：精准路由）
     */
    @Bean
    public DirectExchange directExchange() {
        // durable: 持久化（重启RabbitMQ不丢失）
        // autoDelete: 自动删除（无队列绑定时删除，企业级一般设为false）
        return new DirectExchange(directExchange, true, false);
    }

    /**
     * 2. 声明主题交换机（企业级常用：模糊路由）
     */
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(topicExchange, true, false);
    }

    /**
     * 3. 声明普通队列（绑定死信配置，企业级防消息丢失）
     */
    @Bean
    public Queue normalQueue() {
        return QueueBuilder.durable(normalQueue)
                // 配置死信交换机
                .deadLetterExchange(directExchange)
                // 配置死信路由键
                .deadLetterRoutingKey(deadRoutingKey)
                // 队列消息过期时间（30秒，可选）
                .ttl(30000)
                .build();
    }

    /**
     * 4. 声明死信队列（企业级必配：处理消费失败的消息）
     */
    @Bean
    public Queue deadQueue() {
        return QueueBuilder.durable(deadQueue).build();
    }

    /**
     * 5. 绑定普通队列到直连交换机
     */
    @Bean
    public Binding normalQueueBinding() {
        return BindingBuilder.bind(normalQueue()).to(directExchange()).with(normalRoutingKey);
    }

    /**
     * 6. 绑定死信队列到直连交换机
     */
    @Bean
    public Binding deadQueueBinding() {
        return BindingBuilder.bind(deadQueue()).to(directExchange()).with(deadRoutingKey);
    }

    /**
     * 7. 消息转换器（企业级：JSON格式序列化，避免序列化问题）
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 8. 配置RabbitTemplate（企业级：消息发送确认、返回确认）
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter());

        // 开启消息发送确认（确认消息是否到达交换机）
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                // 企业级处理：记录日志、告警、重试
                System.err.println("消息发送到交换机失败：" + cause);
            }
        });

        // 开启消息返回确认（确认消息是否到达队列）
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnsCallback(returnedMessage -> {
            // 企业级处理：记录日志、告警、重试
            System.err.println("消息发送到队列失败：" + returnedMessage.getReplyText());
        });

        return rabbitTemplate;
    }
}

4. 生产者实现（企业级封装）

4.1 消息发送工具类（通用封装）

在 cc.hep.util 包下创建 RabbitMQProducerUtil，封装通用的消息发送方法（企业级：避免重复代码，统一处理发送逻辑）：

package cc.hep.util;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * RabbitMQ 生产者工具类（企业级通用封装）
 */
@Component
public class RabbitMQProducerUtil {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MessageConverter messageConverter;

    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 发送普通消息（通用方法）
     * @param exchange 交换机
     * @param routingKey 路由键
     * @param message 消息内容（任意对象）
     */
    public void sendMessage(String exchange, String routingKey, Object message) {
        try {
            // 企业级：设置唯一消息ID（用于幂等、追踪）
            String messageId = UUID.randomUUID().toString();
            org.springframework.amqp.core.Message amqpMessage = messageConverter.toMessage(message, new org.springframework.amqp.core.MessageProperties());
            amqpMessage.getMessageProperties().setMessageId(messageId);
            // 发送消息
            rabbitTemplate.send(exchange, routingKey, amqpMessage);
            // 使用 Jackson 序列化消息内容
            String jsonContent = objectMapper.writeValueAsString(message);
            System.out.println("消息发送成功，ID：" + messageId + "，内容：" + jsonContent);
        } catch (Exception e) {
            // 企业级处理：记录异常、重试、告警
            System.err.println("消息发送失败：" + e.getMessage());
            throw new RuntimeException("消息发送失败", e);
        }
    }

    /**
     * 发送延迟消息（基于死信队列实现，企业级常用）
     * @param exchange 交换机
     * @param routingKey 路由键
     * @param message 消息内容
     * @param delayTime 延迟时间（毫秒）
     */
    public void sendDelayMessage(String exchange, String routingKey, Object message, long delayTime) {
        try {
            String messageId = UUID.randomUUID().toString();
            org.springframework.amqp.core.MessageProperties properties = new org.springframework.amqp.core.MessageProperties();
            properties.setMessageId(messageId);
            // 设置消息延迟时间（覆盖队列的默认TTL）
            properties.setExpiration(String.valueOf(delayTime));
            org.springframework.amqp.core.Message amqpMessage = messageConverter.toMessage(message, properties);
            // 发送延迟消息
            rabbitTemplate.send(exchange, routingKey, amqpMessage);
            // 使用 Jackson 序列化消息内容
            String jsonContent = objectMapper.writeValueAsString(message);
            System.out.println("延迟消息发送成功，ID：" + messageId + "，延迟时间：" + delayTime + "ms，内容：" + jsonContent);
        } catch (Exception e) {
            System.err.println("延迟消息发送失败：" + e.getMessage());
            throw new RuntimeException("延迟消息发送失败", e);
        }
    }
}

4.2 业务层生产者（模拟企业业务）

在 cc.hep.service 包下创建 RabbitMQProducerService，封装业务相关的消息发送逻辑：

package cc.hep.service;

import cc.hep.util.RabbitMQProducerUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

/**
 * 消息生产者业务层（模拟企业级业务场景）
 */
@Service
public class RabbitMQProducerService {

    @Autowired
    private RabbitMQProducerUtil rabbitMQProducerUtil;

    // 从配置文件读取常量
    @Value("${rabbitmq.exchange.direct}")
    private String directExchange;

    @Value("${rabbitmq.routing-key.normal}")
    private String normalRoutingKey;

    /**
     * 发送普通业务消息（例如：订单创建消息）
     * @param orderId 订单ID
     * @param orderName 订单名称
     */
    public void sendOrderMessage(String orderId, String orderName) {
        // 模拟订单对象（企业级：使用实体类）
        OrderMessage orderMessage = new OrderMessage();
        orderMessage.setOrderId(orderId);
        orderMessage.setOrderName(orderName);
        orderMessage.setCreateTime(System.currentTimeMillis());
        // 发送消息
        rabbitMQProducerUtil.sendMessage(directExchange, normalRoutingKey, orderMessage);
    }

    /**
     * 发送延迟业务消息（例如：订单超时未支付提醒）
     * @param orderId 订单ID
     * @param delayTime 延迟时间（毫秒）
     */
    public void sendDelayOrderMessage(String orderId, long delayTime) {
        DelayOrderMessage delayOrderMessage = new DelayOrderMessage();
        delayOrderMessage.setOrderId(orderId);
        delayOrderMessage.setRemindContent("订单超时未支付，请及时处理");
        // 发送延迟消息
        rabbitMQProducerUtil.sendDelayMessage(directExchange, normalRoutingKey, delayOrderMessage, delayTime);
    }

    // 内部静态类：订单消息实体（企业级：单独创建实体类）
    public static class OrderMessage {
        private String orderId;
        private String orderName;
        private long createTime;

        // getter/setter
        public String getOrderId() { return orderId; }
        public void setOrderId(String orderId) { this.orderId = orderId; }
        public String getOrderName() { return orderName; }
        public void setOrderName(String orderName) { this.orderName = orderName; }
        public long getCreateTime() { return createTime; }
        public void setCreateTime(long createTime) { this.createTime = createTime; }
    }

    // 内部静态类：延迟订单消息实体
    public static class DelayOrderMessage {
        private String orderId;
        private String remindContent;

        // getter/setter
        public String getOrderId() { return orderId; }
        public void setOrderId(String orderId) { this.orderId = orderId; }
        public String getRemindContent() { return remindContent; }
        public void setRemindContent(String remindContent) { this.remindContent = remindContent; }
    }
}

4.3 控制器（提供接口测试）

在 cc.hep.controller 包下创建 RabbitMQController，提供 HTTP 接口供外部调用（模拟企业级接口调用场景）：

package cc.hep.controller;

import cc.hep.service.RabbitMQProducerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * RabbitMQ 测试控制器（企业级：提供接口供外部系统调用）
 */
@RestController
public class RabbitMQController {

    @Autowired
    private RabbitMQProducerService rabbitMQProducerService;

    /**
     * 发送普通订单消息
     * @param orderId 订单ID
     * @param orderName 订单名称
     * @return 响应结果
     */
    @GetMapping("/send/order")
    public String sendOrderMessage(@RequestParam String orderId, @RequestParam String orderName) {
        try {
            rabbitMQProducerService.sendOrderMessage(orderId, orderName);
            return "普通订单消息发送成功，订单ID：" + orderId;
        } catch (Exception e) {
            return "普通订单消息发送失败：" + e.getMessage();
        }
    }

    /**
     * 发送延迟订单消息
     * @param orderId 订单ID
     * @param delayTime 延迟时间（毫秒）
     * @return 响应结果
     */
    @GetMapping("/send/delay/order")
    public String sendDelayOrderMessage(@RequestParam String orderId, @RequestParam long delayTime) {
        try {
            rabbitMQProducerService.sendDelayOrderMessage(orderId, delayTime);
            return "延迟订单消息发送成功，订单ID：" + orderId + "，延迟时间：" + delayTime + "ms";
        } catch (Exception e) {
            return "延迟订单消息发送失败：" + e.getMessage();
        }
    }
}

5. 消费者实现（企业级消费策略）

5.1 普通消息消费者（手动ACK）

在 cc.hep.consumer 包下创建 NormalMessageConsumer，处理普通队列的消息（企业级：手动ACK，异常处理，重试机制）：

package cc.hep.consumer;

import cc.hep.service.RabbitMQProducerService;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

/**
 * 普通消息消费者（企业级：手动ACK，异常处理）
 */
@Component
public class NormalMessageConsumer {

    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 消费普通队列消息
     * @param message 消息体
     * @param channel 通道（用于手动ACK）
     * @param deliveryTag 消息投递标签（唯一标识）
     * @throws Exception 异常
     */
    @RabbitListener(queues = "${rabbitmq.queue.normal}")
    public void consumeNormalMessage(Message message, Channel channel,
                                     @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws Exception {
        try {
            // 1. 获取消息ID和内容
            String messageId = message.getMessageProperties().getMessageId();
            String messageContent = new String(message.getBody(), "UTF-8");
            System.out.println("收到普通队列消息，ID：" + messageId + "，内容：" + messageContent);

            // 2. 使用 Jackson 解析消息（企业级：校验消息格式、幂等性）
            RabbitMQProducerService.OrderMessage orderMessage =
                    objectMapper.readValue(messageContent, RabbitMQProducerService.OrderMessage.class);

            // 3. 模拟业务处理（例如：更新订单状态、发送通知）
            System.out.println("处理订单消息，订单ID：" + orderMessage.getOrderId() +
                    "，订单名称：" + orderMessage.getOrderName());

            // 4. 手动确认消息（ACK）：消息处理成功，告知RabbitMQ删除消息
            channel.basicAck(deliveryTag, false);
            System.out.println("消息处理成功，已ACK，消息ID：" + messageId);

        } catch (Exception e) {
            System.err.println("消息处理失败，消息ID：" + message.getMessageProperties().getMessageId() +
                    "，异常：" + e.getMessage());

            // 企业级处理策略：
            // 1. 重试次数未达上限：拒绝消息并重新入队（basicNack + requeue=true）
            // 2. 重试次数已达上限：拒绝消息并进入死信队列（basicNack + requeue=false）
            Integer retryCount = message.getMessageProperties().getHeader("x-retry-count");
            if (retryCount == null) {
                retryCount = 1;
            } else {
                retryCount++;
            }

            if (retryCount <= 3) {
                // 重新入队，继续重试
                channel.basicNack(deliveryTag, false, true);
                System.out.println("消息重试中，当前重试次数：" + retryCount +
                        "，消息ID：" + message.getMessageProperties().getMessageId());
            } else {
                // 拒绝入队，进入死信队列
                channel.basicNack(deliveryTag, false, false);
                System.out.println("消息重试次数已达上限，进入死信队列，消息ID：" +
                        message.getMessageProperties().getMessageId());
            }
        }
    }
}

5.2 死信消息消费者（处理失败消息）

在 cc.hep.consumer 包下创建 DeadLetterConsumer，处理死信队列的消息（企业级：记录失败消息，人工介入处理）：

package cc.hep.consumer;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

/**
 * 死信消息消费者（企业级：处理消费失败的消息）
 */
@Component
public class DeadLetterConsumer {

    /**
     * 消费死信队列消息
     * @param message 消息体
     * @param channel 通道
     * @param deliveryTag 投递标签
     * @throws Exception 异常
     */
    @RabbitListener(queues = "${rabbitmq.queue.dead}")
    public void consumeDeadLetterMessage(Message message, Channel channel,
                                         @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws Exception {
        try {
            // 1. 获取消息信息
            String messageId = message.getMessageProperties().getMessageId();
            String messageContent = new String(message.getBody(), "UTF-8");
            System.err.println("收到死信队列消息，ID：" + messageId + "，内容：" + messageContent);

            // 2. 企业级处理：记录死信消息到数据库/日志，通知运维人员人工介入
            System.err.println("死信消息已记录，等待人工处理，消息ID：" + messageId);

            // 3. 手动ACK（死信消息处理完成）
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            System.err.println("死信消息处理失败：" + e.getMessage());
            // 死信消息处理失败，直接拒绝（避免无限循环）
            channel.basicNack(deliveryTag, false, false);
        }
    }
}

7. 项目启动与验证

7.1 启动项目

1. 找到项目主类 SpringBootRabbitMqApplication（cc.hep 包下），右键点击 Run。
2. 控制台无报错，且显示 Started SpringBootRabbitMqApplication in xxx seconds，说明项目启动成功。

7.2 接口测试

1. 发送普通订单消息：访问 http://localhost:8080/send/order?orderId=ORDER_001&orderName=苹果16 Pro，返回“普通订单消息发送成功”。
2. 发送延迟订单消息：访问 http://localhost:8080/send/delay/order?orderId=ORDER_002&delayTime=5000，返回“延迟订单消息发送成功”。
3. 查看控制台：能看到消费者打印的消息处理日志，验证消息发送和消费流程正常。

7.3 RabbitMQ 管理界面验证

1. 访问 http://localhost:15672，登录后进入 Queues 页面。
2. 查看 hep.normal.queue 和 hep.dead.queue 的消息数量：
   • 普通消息发送后，hep.normal.queue 消息数先增加，消费后变为0。
   • 延迟消息发送后，hep.normal.queue 消息数增加，延迟时间到后变为0，hep.dead.queue 无消息（若处理正常）。

8. 企业级生产环境注意事项

8.1 性能优化

1. 连接池配置：根据业务量调整 spring.rabbitmq.cache.connection.size 和 spring.rabbitmq.cache.channel.size，避免连接耗尽。
2. 消费并发数：根据服务器CPU核心数调整 spring.rabbitmq.listener.simple.concurrency 和 max-concurrency，建议CPU核心数*2。
3. 预取数：prefetch 不宜过大（避免消费堆积），也不宜过小（降低消费效率），建议10-50。

8.2 可靠性保障

1. 消息持久化：交换机、队列、消息均需设置持久化（代码中已配置 durable=true）。
2. 发送确认：开启 confirmCallback 和 returnsCallback，确保消息到达交换机和队列。
3. 手动ACK：消费端必须使用手动ACK，避免消息丢失。
4. 死信队列：所有业务队列必须绑定死信队列，处理消费失败的消息。

8.3 监控与告警

1. 集成 Prometheus + Grafana 监控 RabbitMQ 指标（消息堆积、消费速率、连接数等）。
2. 配置告警规则：消息堆积超过阈值、消费失败次数过多时，及时告警给运维人员。

8.4 幂等性保障

1. 基于消息ID + Redis 实现幂等：消费前先检查Redis中是否有该消息ID，有则跳过，无则处理并记录。
2. 业务层幂等：例如订单消息处理时，先检查订单状态，避免重复处理。

总结

核心要点回顾

1. 项目搭建：基于 IDEA 初始化 SpringBoot 项目，引入 RabbitMQ 核心依赖，配置文件对齐企业级规范（连接池、手动ACK、重试机制）。
2. 核心配置：通过配置类声明交换机、队列、绑定关系，配置 RabbitTemplate 实现消息发送确认和JSON序列化。
3. 生产者：封装通用发送工具类，业务层结合实际场景发送普通/延迟消息，控制器提供HTTP接口。
4. 消费者：手动ACK处理消息，异常时根据重试次数决定重新入队或进入死信队列，死信队列消费失败消息供人工处理。
5. 企业级保障：消息持久化、发送确认、手动ACK、死信队列是保障RabbitMQ可靠性的核心，生产环境需重点关注。

扩展方向

1. 集成 RocketMQ/ Kafka：若业务量极大，可替换为更高吞吐量的消息中间件。
2. 消息追踪：开启 RabbitMQ 的 rabbitmq_tracing 插件，追踪消息全链路。
3. 分布式事务：结合 RabbitMQ 实现最终一致性分布式事务（Seata + RabbitMQ）。