消息队列 RabbitMQ

为什么需要消息队列

消息队列（Message Queue）是分布式系统中解决异步通信问题的核心中间件，主要解决三类场景：

核心概念名词解释

Exchange 类型

消息从发送到消费的完整链路

Spring Boot 集成 RabbitMQ

依赖配置

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

Queue、Exchange、Binding 配置

@Configuration
public class RabbitConfig {

    public static final String ORDER_QUEUE = "order.created.queue";
    public static final String ORDER_EXCHANGE = "order.events";
    public static final String ORDER_ROUTING_KEY = "order.created";
    public static final String DLQ_NAME = "order.dead.letter.queue";
    public static final String DLX_NAME = "order.dead.letter.exchange";

    // 死信交换器
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange(DLX_NAME);
    }

    // 死信队列
    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable(DLQ_NAME).build();
    }

    @Bean
    public Binding dlqBinding() {
        return BindingBuilder.bind(deadLetterQueue())
            .to(deadLetterExchange()).with("order.created");
    }

    // 业务队列（绑定死信配置）
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE)
            .deadLetterExchange(DLX_NAME)      // 死信路由到 DLX
            .deadLetterRoutingKey("order.created")
            .ttl(60000)                          // 消息 TTL 60秒
            .build();
    }

    @Bean
    public TopicExchange orderExchange() {
        return new TopicExchange(ORDER_EXCHANGE);
    }

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

生产者：发送消息

@Service
@RequiredArgsConstructor
public class OrderService {

    private final RabbitTemplate rabbitTemplate;
    private final OrderRepository orderRepository;

    @Transactional
    public Order createOrder(CreateOrderRequest req) {
        Order order = /* 创建订单并保存 */ orderRepository.save(buildOrder(req));

        // 发送消息到 RabbitMQ（事务提交后发送更安全）
        OrderCreatedEvent event = new OrderCreatedEvent(
            order.getId(), order.getUserId(), order.getTotalAmount()
        );
        rabbitTemplate.convertAndSend(
            RabbitConfig.ORDER_EXCHANGE,
            RabbitConfig.ORDER_ROUTING_KEY,
            event
        );
        return order;
    }
}

消费者：处理消息（手动 ACK）

@Component
public class OrderCreatedConsumer {

    @RabbitListener(queues = RabbitConfig.ORDER_QUEUE, ackMode = "MANUAL")
    public void handleOrderCreated(
            @Payload OrderCreatedEvent event,
            Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        try {
            // 处理业务（幂等设计：用 orderId 去重）
            processOrderCreated(event);

            // 手动 ACK：通知 RabbitMQ 消息已成功处理
            channel.basicAck(tag, false);

        } catch (BusinessException e) {
            // 业务异常：不重试，直接 nack 进入死信队列
            log.error("业务处理失败，进入死信队列: {}", event, e);
            channel.basicNack(tag, false, false);  // requeue=false → 死信

        } catch (Exception e) {
            // 临时异常（网络、超时）：重新入队重试
            log.warn("临时异常，重新入队: {}", event, e);
            channel.basicNack(tag, false, true);   // requeue=true → 重试
        }
    }
}