RabbitMQ 从零到实战：概念、配置与 Spring Boot 集成实践指南

Producer(生产者)
    │
    │ 发送消息
    ▼
Exchange(交换机) ──── Binding(绑定规则) ──── Queue(队列)
                                                │
                                                │ 消费消息
                                                ▼
                                         Consumer(消费者)

docker run -d --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=admin123 \
  rabbitmq:3-management

1. 创建 Exchange
   - Name: my-app.exchange
   - Type: direct
   - Durable: true
2. 创建 Queue
   - Name: my-app.import-task
   - Durable: true
3. 创建 Binding
   - Source: my-app.exchange
   - Destination: my-app.import-task
   - Routing Key: import-task（通常和 Queue 名相同）

spring:
  rabbitmq:
    # AMQP 接入点地址（阿里云格式，自建则为 host:port）
    addresses: amqp-cn-xxx5.mq-amqp.cn-qingdao-xxxx-a.aliyuncs.com
    # 认证信息（阿里云 AMQP 用 Base64 编码的 AccessKey）
    username: xxxx...（Base64编码）
    password: xxxxxx...（Base64编码）
    # 阿里云 AMQP 特有配置
    access-key-id: xxxxxx
    access-key-secret: xxxx
    instance-id: amqp-cn-xxxx
    region: cn-qingdao
    # 消费者监听配置
    listener:
      simple:
        # 消费失败时不重新入队（避免死循环）
        default-requeue-rejected: false
        # 最小并发消费者数
        concurrency: 1
        # 最大并发消费者数
        max-concurrency: 3
        # 预取数量（每次从 Broker 拉取几条未确认的消息）
        prefetch: 1

spring:
  rabbitmq:
    host: 192.168.1.1xx
    port: 5672
    username: myapp
    password: myapp123
    virtual-host: /myapp
    connection-timeout: 5000
    listener:
      simple:
        acknowledge-mode: manual      # 手动确认
        concurrency: 2
        max-concurrency: 5
        prefetch: 1
        default-requeue-rejected: false
    publisher-confirm-type: correlated  # 生产者确认
    publisher-returns: true             # 消息无法路由时回调

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

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: admin123
    virtual-host: /
app:
  mq:
    exchange: app.exchange
    queue: app.import-task
    routing-key: import-task

package com.example.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * RabbitMQ 资源配置.
 * 应用启动时自动创建 Exchange、Queue 并建立绑定关系.
 */
@Configuration
public class RabbitMqConfig {
  @Value("${app.mq.exchange}")
  private String exchangeName;
  @Value("${app.mq.queue}")
  private String queueName;
  @Value("${app.mq.routing-key}")
  private String routingKey;
  /** 声明持久化交换机. */
  @Bean
  public DirectExchange importExchange() {
    return new DirectExchange(exchangeName, true, false);
  }
  /** 声明持久化队列. */
  @Bean
  public Queue importQueue() {
    return new Queue(queueName, true);
  }
  /** 建立绑定关系. */
  @Bean
  public Binding importBinding(Queue importQueue, DirectExchange importExchange) {
    return BindingBuilder.bind(importQueue).to(importExchange).with(routingKey);
  }
}

package com.example.dto;
import java.io.Serializable;
import java.util.List;
import lombok.Data;
/**
 * 导入任务消息体.
 */
@Data
public class ImportTaskMessage implements Serializable {
  /** 批次ID. */
  private Long batchId;
  /** 批次号. */
  private String batchNo;
  /** 待插入的数据（JSON序列化后体积较大时，可改为只传batchId，消费端自己查数据）. */
  private List<ImportRecord> records;
}

package com.example.mq;
import com.example.dto.ImportTaskMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import jakarta.annotation.Resource;
/**
 * 导入任务消息生产者.
 */
@Slf4j
@Component
public class ImportTaskProducer {
  @Resource
  private RabbitTemplate rabbitTemplate;
  @Value("${app.mq.exchange}")
  private String exchange;
  @Value("${app.mq.routing-key}")
  private String routingKey;
  /**
   * 发送导入任务消息.
   */
  public void send(ImportTaskMessage message) {
    log.info("发送导入任务消息，批次号:{}", message.getBatchNo());
    rabbitTemplate.convertAndSend(exchange, routingKey, message);
    log.info("导入任务消息发送完成");
  }
}

package com.example.mq;
import com.example.dto.ImportRecord;
import com.example.dto.ImportTaskMessage;
import com.example.mapper.ImportRecordMapper;
import jakarta.annotation.Resource;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
 * 导入任务消息消费者.
 */
@Slf4j
@Component
public class ImportTaskConsumer {
  private static final int BATCH_SIZE = 2000;
  @Resource
  private ImportRecordMapper importRecordMapper;
  /**
   * 消费导入任务.
   * 使用 @RabbitListener 注解声明监听哪个队列.
   */
  @RabbitListener(queues = "${app.mq.queue}")
  public void consume(ImportTaskMessage message) {
    log.info("收到导入任务，批次号:{}，数据量:{}", 
        message.getBatchNo(), message.getRecords().size());
    try {
      List<ImportRecord> records = message.getRecords();
      int total = records.size();
      // 分批插入
      for (int i = 0; i < total; i += BATCH_SIZE) {
        int end = Math.min(i + BATCH_SIZE, total);
        List<ImportRecord> batch = records.subList(i, end);
        importRecordMapper.batchInsert(batch);
      }
      log.info("导入任务处理完成，批次号:{}，共插入{}条", message.getBatchNo(), total);
    } catch (Exception e) {
      log.error("导入任务处理失败，批次号:{}", message.getBatchNo(), e);
      // 抛出异常后，根据 default-requeue-rejected 配置决定是否重新入队
      throw e;
    }
  }
}

package com.example.service.impl;
import com.example.dto.ImportRecord;
import com.example.dto.ImportTaskMessage;
import com.example.mq.ImportTaskProducer;
import com.example.service.ImportService;
import jakarta.annotation.Resource;
import java.util.List;
import org.springframework.stereotype.Service;
@Service
public class ImportServiceImpl implements ImportService {
  @Resource
  private ImportTaskProducer producer;
  @Override
  public String importData(List<ImportRecord> records, String operatorId) {
    // 同步校验
    for (int i = 0; i < records.size(); i++) {
      if (records.get(i).getAmount() <= 0) {
        throw new RuntimeException("第" + (i + 1) + "行：数量必须大于0");
      }
    }
    // 保存主表...（略）
    String batchNo = "20260610001";
    // 发送MQ消息，异步执行插入
    ImportTaskMessage message = new ImportTaskMessage();
    message.setBatchId(1L);
    message.setBatchNo(batchNo);
    message.setRecords(records);
    producer.send(message);
    return "导入任务已提交，批次号：" + batchNo;
  }
}

请求线程                      RabbitMQ Broker              消费者线程
   │                              │                          │
   │── 同步校验 ──→                │                          │
   │── 保存主表 ──→                │                          │
   │── producer.send(msg) ──→     │                          │
   │                              │── 持久化消息 ──→           │
   │← 返回"任务已提交" ──          │                          │
   │                              │── 推送消息 ──→            │
   │  (HTTP已响应)                 │                          │── 批量INSERT
   │                              │                          │── ...
   │                              │                          │── INSERT完成
   │                              │←── ACK ──                │
   │                              │── 删除消息               │

Producer ──确认──→ Exchange ──确认──→ Queue ──确认──→ Consumer
   │                  │                │               │
   │ publisher-confirm │ mandatory     │ 持久化        │ 手动ACK

正常 Queue ──(消费失败/超时/被拒绝)──→ 死信 Exchange ──→ 死信 Queue