RabbitMQ 入门：交换机与队列

消息队列不是“发个消息”这么简单

订单创建后要通知库存、积分、短信，同步调用越串越长，任何一个下游抖动都会拖慢下单。RabbitMQ 的思路是把“必须现在做”和“可以稍后做”拆开：下单只负责把消息可靠地交给 Broker，下游按自己的节奏消费。代价是系统从“调用成功即完成”变成了“最终一致”，消息可能重复、可能延迟，也可能进死信队列。

用好 RabbitMQ 的前提，是先把交换机、队列、绑定这三者的关系理顺。很多人卡在“消息发出去了但消费者没收到”，最后发现是 routing key 拼错，或者交换机类型选错。

交换机、队列与绑定

生产者不直接把消息发给队列，而是发给交换机；交换机根据类型和 routing key 把消息路由到一个或多个队列；队列再推给消费者。绑定（Binding）就是交换机和队列之间的路由规则。

默认交换机 "" 是一个 direct 交换机，用队列名当 routing key 可以直接投递，适合简单场景，但生产环境更建议显式声明交换机和绑定，避免队列名变更导致消息丢失。

代码：订单创建消息

Spring Boot 3.2 使用 spring-boot-starter-amqp，配置里打开发布确认和手动 ack：

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    publisher-confirm-type: correlated
    publisher-returns: true
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 20

声明交换机、队列和绑定：

@Configuration
public class RabbitConfig {

    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_QUEUE = "order.created.queue";
    public static final String ORDER_ROUTING_KEY = "order.created";

    @Bean
    public DirectExchange orderExchange() {
        return ExchangeBuilder.directExchange(ORDER_EXCHANGE).durable(true).build();
    }

    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE).build();
    }

    @Bean
    public Binding orderBinding(Queue orderQueue, DirectExchange orderExchange) {
        return BindingBuilder.bind(orderQueue).to(orderExchange).with(ORDER_ROUTING_KEY);
    }
}

发送与消费：

@Service
public class OrderMessageProducer {

    private final RabbitTemplate rabbitTemplate;

    public OrderMessageProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void send(OrderCreatedMessage message) {
        rabbitTemplate.convertAndSend(
                RabbitConfig.ORDER_EXCHANGE,
                RabbitConfig.ORDER_ROUTING_KEY,
                message,
                new CorrelationData(message.orderId().toString()));
    }
}

@Component
public class OrderCreatedConsumer {

    private static final Logger log = LoggerFactory.getLogger(OrderCreatedConsumer.class);

    @RabbitListener(queues = RabbitConfig.ORDER_QUEUE)
    public void onMessage(OrderCreatedMessage message,
                          Channel channel,
                          @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        try {
            // 业务处理必须幂等，重复消息可能再次到达
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理订单消息失败, orderId={}", message.orderId(), e);
            channel.basicNack(deliveryTag, false, false);
        }
    }
}

basicNack 的第三个参数是 requeue。传 false 表示不重新入队，消息会进入死信交换机（如果配置了）或被丢弃；传 true 会反复投递，处理不好就是死循环。

设计上有几个原则。按业务域划分交换机，比如 order.exchange、inventory.exchange，不要所有业务共用一个。队列名带上消费者语义，例如 order.created.point.queue，这样看到队列名就知道谁在消费。一个队列有多个消费者时，消息是竞争消费，每条消息只会被其中一个消费者拿到；需要广播时，给每个消费者建独立队列并绑定到 fanout 交换机。

队列要声明为 durable，消息要设置持久化，否则 Broker 重启后队列和消息都可能消失。生产环境还建议使用 quorum 队列，基于 Raft 复制，主节点故障时不会丢已确认的消息，代价是吞吐略低、对磁盘和内存要求更高。

常见坑

1. 消息发出去了，但没有进队列

交换机存在、routing key 不匹配时，消息会被静默丢弃。只开 publisher confirm 不够，它只确认 Broker 收到了消息，不保证路由成功。要同时开启 publisher-returns 并设置 mandatory=true，在 ReturnsCallback 里记录无法路由的消息。

2. durable 和 persistent 只设了一个

队列 durable 只保证队列元数据在重启后还在，消息本身还要 MessageDeliveryMode.PERSISTENT。两者都设，消息才可能在 Broker 重启后保留。即使用了 quorum 队列，也要确认发送端没有把 delivery mode 改成非持久化。

3. 手动 ack 忘了确认

消费方法抛异常前没 ack 也没 nack，消息会一直处于 unacked 状态；prefetch 满了之后，这个消费者再也收不到新消息。表现是队列看起来有消费者，但消息不流动。用 try-catch-finally 确保每条消息都有 ack 或 nack，排查时先看 unacked 数量，再对照消费方法的 ack 分支，基本能定位到漏确认的位置。

4. 消费逻辑不幂等

网络抖动、消费者重启都会导致消息重复投递，RabbitMQ 只保证 at-least-once。用订单号做唯一索引、Redis 记录消费状态、或者数据库乐观锁，都是常见做法。不要指望“消息只来一次”。

5. 反序列化失败

默认的 SimpleMessageConverter 对 POJO 不友好，通常要配置 Jackson2JsonMessageConverter。如果消费端类结构变了，旧消息可能反序列化失败，进而无限重试。给消息加版本号，或者让死信队列接管无法处理的消息。

排查步骤

先看队列有没有积压、消费者数量是否正常：

rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers
rabbitmqctl list_bindings
rabbitmq-diagnostics check_running
rabbitmq-diagnostics check_port_connectivity

messages_ready 高说明没有消费者或消费太慢；messages_unacknowledged 高说明消费者拿了消息但不确认，通常是业务卡住或忘了 ack。再去管理台看连接和 channel 是否频繁重建，网络抖动和心跳超时也会表现成消费停滞。

小结

RabbitMQ 的入门门槛不在 API，而在模型：交换机负责路由，队列负责存储，绑定负责规则，ack 负责可靠性。先把这四件事对应到业务上，再去调 prefetch、确认模式和死信队列，顺序反了只会越调越乱。

参考

• RabbitMQ 官方文档
• 博客园：RabbitMQ 消息丢失怎么解决
• RocketMQ 官方文档
• CSDN：RocketMQ / RabbitMQ 深度解析
• Elasticsearch 官方指南
• 博客园：ES 倒排索引实战
• 博客园：Kafka 消费者组订阅问题