Spring AMQP 是 Spring 官方提供的对 AMQP 协议（以 RabbitMQ 为主要实现）的抽象封装，它把底层的连接管理、信道操作、序列化等繁琐细节统一到模板类和注解驱动模型中。在 Spring Boot 项目中使用 Spring AMQP，核心目标往往不只是收发消息，而是借助其高级特性构建可靠的消息管道——比如保证消息不丢失、处理消费失败后的重新投递、实现延迟队列等。本文会从实际工程角度出发，梳理这些特性的配置方法和运行机制。

环境准备与基础配置

首先在 Maven 或 Gradle 项目中引入 spring-boot-starter-amqp 依赖，它会自动添加 spring-rabbit 和 amqp-client。配置文件里通常只需指定 RabbitMQ 的连接地址、用户名密码以及虚拟主机。例如在 application.yml 中：

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    publisher-confirm-type: correlated
    publisher-returns: true

这里 publisher-confirm-type 设置为 correlated 表示开启生产者确认，publisher-returns 开启失败回退。连接工厂 ConnectionFactory 会自动根据配置创建，RabbitTemplate 和 RabbitAdmin 也会自动装配。如果使用默认的简单队列声明方式，可以在配置类中通过 Queue、Exchange、Binding 三个 Bean 完成拓扑定义。例如声明一个直连交换机和一个持久化队列：

@Configuration
public class RabbitConfig {
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable("order.queue")
                .withArgument("x-message-ttl", 60000)
                .build();
    }
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange("order.exchange", true, false);
    }
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue())
                .to(orderExchange())
                .with("order.key");
    }
}

上面代码中 QueueBuilder.durable 创建持久化队列，并设置了 x-message-ttl 参数表示队列中消息的存活时间为 60 秒。生产端使用 RabbitTemplate.convertAndSend 发送消息时，消息会路由到 order.queue。消费端则通过 @RabbitListener 注解监听队列，方法参数可以接收消息体对象，Spring 会自动完成转换。

消息可靠性保障：生产者确认与消费者手动确认

默认情况下，生产者发送消息后不会知道消息是否成功到达 Broker。开启 publisher-confirm-type: correlated 后，RabbitTemplate 可以注册 ConfirmCallback 来接收确认结果。同时开启 publisher-returns 并设置 mandatory 为 true 时，当消息无法路由到任何队列，Broker 会返回该消息，此时可以注册 ReturnsCallback 处理。下面是一个发送并设置回调的示例：

@Service
public class OrderMessageSender {
    private final RabbitTemplate rabbitTemplate;
    public OrderMessageSender(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        this.rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                System.out.println("消息确认成功: " + correlationData.getId());
            } else {
                System.out.println("消息确认失败: " + cause);
            }
        });
        this.rabbitTemplate.setReturnsCallback(returned -> {
            System.out.println("消息路由失败: " + returned.getMessage());
        });
    }
    public void sendOrder(String orderId) {
        CorrelationData correlationData = new CorrelationData(orderId);
        rabbitTemplate.convertAndSend("order.exchange", "order.key",
                "订单消息内容", correlationData);
    }
}

生产者确认只能保证消息到达交换机，并不保证消息被队列持久化。如果需要更强的保证，可以结合事务或发送方确认机制。消费者端默认是自动确认（autoAck），即消息一旦投递给监听方法就立即 ack。如果监听方法抛出异常，消息会被丢弃或重新入队？在自动确认模式下，异常会导致消息被拒绝并重新入队（取决于配置）。为了精确控制，建议使用手动确认模式。在 application.yml 中配置：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual

然后在监听方法中接收 Channel 和 Message 参数，业务处理成功后调用 basicAck，失败则调用 basicNack 或 basicReject。这样能确保只有成功处理的消息才会从队列中移除。例如：

@RabbitListener(queues = "order.queue")
public void handleOrder(Message message, Channel channel) throws IOException {
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    try {
        String body = new String(message.getBody());
        System.out.println("处理订单: " + body);
        // 业务逻辑
        channel.basicAck(deliveryTag, false);
    } catch (Exception e) {
        channel.basicNack(deliveryTag, false, true); // 重新入队
    }
}

手动确认模式需要特别注意：如果忘记确认或拒绝，消息会一直处于 unacked 状态，当连接断开后会重新回到队列，可能导致重复消费。因此无论是 ack 还是 nack，都必须在 finally 或保证逻辑闭环。

死信队列与延迟消息实现

死信队列（Dead Letter Queue，DLQ）是处理无法被正常消费的消息的通用方案。当消息被拒绝（basicReject 或 basicNack 且 requeue=false）、消息过期（TTL 到期）、队列达到最大长度时，消息会被转发到死信交换机。配置死信队列需要在原始队列上设置 x-dead-letter-exchange 和 x-dead-letter-routing-key 参数。下面定义一个业务队列及其死信队列：

@Bean
public Queue businessQueue() {
    return QueueBuilder.durable("business.queue")
            .withArgument("x-dead-letter-exchange", "dlx.exchange")
            .withArgument("x-dead-letter-routing-key", "dlx.key")
            .build();
}
@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable("dead.letter.queue").build();
}
@Bean
public DirectExchange dlxExchange() {
    return new DirectExchange("dlx.exchange", true, false);
}
@Bean
public Binding dlxBinding() {
    return BindingBuilder.bind(deadLetterQueue())
            .to(dlxExchange())
            .with("dlx.key");
}

延迟消息通常利用 TTL 和死信队列组合实现：消息先发送到一个设置了 TTL 的队列（或消息本身带 TTL），过期后转发到实际消费队列。也可以使用 RabbitMQ 的延迟消息插件（rabbitmq_delayed_message_exchange），但需要额外安装插件。Spring AMQP 对插件也有支持，使用 CustomExchange 定义类型为 x-delayed-message 的交换机即可。基于 TTL 的方式有个缺点：如果队列中前面的消息过期时间较长，后面的短 TTL 消息会被阻塞，因为队列遵循 FIFO。因此更适合使用消息级别的 TTL，但 RabbitMQ 只有在消息到达队列头部时才会判断是否过期。插件方式则不存在这个问题。下面展示通过 x-delayed-message 实现延迟队列：

@Bean
public CustomExchange delayedExchange() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-delayed-type", "direct");
    return new CustomExchange("delayed.exchange", "x-delayed-message", true, false, args);
}
@Bean
public Queue delayedQueue() {
    return QueueBuilder.durable("delayed.queue").build();
}
@Bean
public Binding delayedBinding() {
    return BindingBuilder.bind(delayedQueue())
            .to(delayedExchange())
            .with("delayed.key")
            .noargs();
}

发送时设置消息头 x-delay 指定延迟毫秒数，例如：

MessageProperties props = new MessageProperties();
props.setHeader("x-delay", 5000);
Message message = new Message("延迟消息".getBytes(), props);
rabbitTemplate.send("delayed.exchange", "delayed.key", message);

死信队列和延迟消息是高级消息队列中非常实用的特性，能够很好地应对订单超时取消、重试退避等场景。

并发消费与消息重试机制

默认情况下，@RabbitListener 容器使用单个线程顺序消费消息。当消息处理耗时较长时，可以增加并发消费者数量来提升吞吐量。在 application.yml 中配置 spring.rabbitmq.listener.simple.concurrency 为固定值，或者使用 min 和 max 配合 max-concurrency 动态调整。例如：

spring:
  rabbitmq:
    listener:
      simple:
        concurrency: 5
        max-concurrency: 10
        prefetch: 1

prefetch 控制每个消费者预取的消息数量，设为 1 可以实现更公平的任务分发，避免某个消费者积压过多消息。对于需要严格顺序消费的场景，需要将并发数设置为 1 并使用单个队列。并发消费与手动确认结合时，要确保 Channel 和 deliveryTag 的操作是线程安全的，因为每个消费者线程持有独立的信道，但同一个线程内的顺序是固定的。

Spring AMQP 的监听容器支持自动重试机制，可以在监听方法抛出异常后由容器内部重试，而不会立即 nack 重新入队。这样既可以避免消息在队列中反复投递，又能在一定次数内处理瞬时故障。配置重试参数如下：

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 1000ms
          multiplier: 2.0
          max-interval: 10000ms

当重试次数耗尽后，如果配置了死信队列，消息会被拒接并进入死信队列；如果没有死信队列，消息会被丢弃或重新入队（取决于 default-requeue-rejected）。为了避免无限重试，建议将 default-requeue-rejected 设置为 false，同时配置死信队列来兜底。此外，Spring Retry 的 BackOffPolicy 可以自定义退避策略，例如指数退避加抖动，防止重试风暴。

整套配置下来，消息系统就能在面对瞬时故障、业务异常和流量高峰时保持稳定。理解每一项配置背后的机制，是构建高级消息队列应用的关键。