RabbitMQ 如何削峰?一文手把手教你扛住流量洪峰(Spring Boot + Java 实战)

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

# application.yml
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        acknowledge-mode: manual   # 👈 手动ACK（关键！）
        prefetch: 5                # 👈 每个消费者最多缓存5条未确认消息（限流核心！）
        concurrency: 10            # 启动10个消费者线程
        max-concurrency: 20

@Configuration
public class PeakShavingConfig {
    public static final String PEAK_QUEUE = "peak.order.queue";
    @Bean
    public Queue peakQueue() {
        return QueueBuilder.durable(PEAK_QUEUE)
                .withArgument("x-max-length", 100000)     // 队列最大长度10万
                .withArgument("x-overflow", "drop-head")  // 超长时丢弃最早消息（防OOM）
                .build();
    }
    @Bean
    public DirectExchange peakExchange() {
        return new DirectExchange("peak.exchange", true, false);
    }
    @Bean
    public Binding peakBinding() {
        return BindingBuilder.bind(peakQueue())
                .to(peakExchange())
                .with("order.create");
    }
}

@Service
public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public String createOrderFast(String userId, String goodsId) {
        // 1. 校验参数（快速失败）
        if (!validate(userId, goodsId)) {
            return "参数错误";
        }
        // 2. 直接发消息到队列（毫秒级响应）
        String orderId = UUID.randomUUID().toString();
        OrderMessage msg = new OrderMessage(orderId, userId, goodsId);
        rabbitTemplate.convertAndSend("peak.exchange", "order.create", msg);
        // 3. 立即返回“请求已接收”，结果异步通知
        return "下单成功，请等待处理结果";
    }
}

@Component
public class OrderConsumer {
    @RabbitListener(queues = PeakShavingConfig.PEAK_QUEUE)
    public void handle(Message message, Channel channel) throws IOException {
        try {
            // 1. 解析消息
            OrderMessage order = parse(message);
            // 2. 执行耗时业务（如扣库存、生成订单）
            processOrder(order); // 假设平均耗时 200ms
            // 3. 手动 ACK（成功才确认）
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            // 4. 失败则拒绝并重新入队（可加重试次数限制）
            channel.basicNack(
                message.getMessageProperties().getDeliveryTag(),
                false,
                true // requeue=true
            );
            log.error("Order processing failed", e);
        }
    }
}

# ❌ 危险配置！
spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto  # 自动ACK
        prefetch: 0             # 无限制拉取

// ❌ 无限制队列
return new Queue("infinite.queue");

rabbitTemplate.setConfirmCallback(...);
rabbitTemplate.setReturnCallback(...);