RabbitMQ在微服务架构中的落地：消息推送 / 解耦 / 削峰填谷

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

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /

@Configuration
public class RabbitMQConfig {
    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_CREATED_QUEUE = "order.created.queue";
    public static final String ORDER_ROUTING_KEY = "order.created";
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange(ORDER_EXCHANGE);
    }
    @Bean
    public Queue orderCreatedQueue() {
        return QueueBuilder.durable(ORDER_CREATED_QUEUE).build();
    }
    @Bean
    public Binding bindingOrderCreated(Queue orderCreatedQueue, DirectExchange orderExchange) {
        return BindingBuilder.bind(orderCreatedQueue)
                .to(orderExchange)
                .with(ORDER_ROUTING_KEY);
    }
}

@Service
public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void createOrder(Order order) {
        // 1. 保存订单到数据库
        orderRepository.save(order);
        // 2. 发布事件（异步）
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            new OrderCreatedEvent(order.getId(), order.getUserId(), order.getAmount())
        );
        // 3. 立即返回，不等待下游处理
    }
}

@Component
public class EmailConsumer {
    @RabbitListener(queues = RabbitMQConfig.ORDER_CREATED_QUEUE)
    public void handleOrderCreated(OrderCreatedEvent event) {
        try {
            // 发送邮件逻辑
            emailService.sendOrderConfirmation(event.getOrderId());
        } catch (Exception e) {
            // 记录日志，可考虑重试或死信队列
            log.error("Failed to send email for order: {}", event.getOrderId(), e);
            throw new AmqpRejectAndDontRequeueException(e); // 避免无限重试
        }
    }
}

public class UserRegisteredEvent {
    private String userId;
    private String email;
    private LocalDateTime registerTime;
    // 构造函数、getter/setter 略
}

@Configuration
public class UserEventConfig {
    public static final String USER_FANOUT_EXCHANGE = "user.fanout.exchange";
    public static final String USER_REGISTERED_QUEUE_POINTS = "user.registered.queue.points";
    public static final String USER_REGISTERED_QUEUE_MARKETING = "user.registered.queue.marketing";
    @Bean
    public FanoutExchange userFanoutExchange() {
        return new FanoutExchange(USER_FANOUT_EXCHANGE);
    }
    @Bean
    public Queue pointsQueue() {
        return QueueBuilder.durable(USER_REGISTERED_QUEUE_POINTS).build();
    }
    @Bean
    public Queue marketingQueue() {
        return QueueBuilder.durable(USER_REGISTERED_QUEUE_MARKETING).build();
    }
    @Bean
    public Binding bindPointsToFanout(Queue pointsQueue, FanoutExchange exchange) {
        return BindingBuilder.bind(pointsQueue).to(exchange);
    }
    @Bean
    public Binding bindMarketingToFanout(Queue marketingQueue, FanoutExchange exchange) {
        return BindingBuilder.bind(marketingQueue).to(exchange);
    }
}

@Service
public class UserService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void registerUser(String email) {
        // 1. 保存用户
        User user = userRepository.save(new User(email));
        // 2. 发布事件（无 Routing Key）
        rabbitTemplate.convertAndSend(
            UserEventConfig.USER_FANOUT_EXCHANGE,
            "", // Fanout 不需要 Routing Key
            new UserRegisteredEvent(user.getId(), email, LocalDateTime.now())
        );
    }
}

@Component
public class PointsConsumer {
    @RabbitListener(queues = UserEventConfig.USER_REGISTERED_QUEUE_POINTS)
    public void onUserRegistered(UserRegisteredEvent event) {
        pointsService.addWelcomePoints(event.getUserId(), 100);
    }
}

@Component
public class MarketingConsumer {
    @RabbitListener(queues = UserEventConfig.USER_REGISTERED_QUEUE_MARKETING)
    public void onUserRegistered(UserRegisteredEvent event) {
        marketingService.sendWelcomeCoupon(event.getEmail());
    }
}

@Configuration
public class SeckillConfig {
    public static final String SECKILL_QUEUE = "seckill.queue";
    @Bean
    public Queue seckillQueue() {
        // 设置队列长度限制，防止内存溢出
        return QueueBuilder.durable(SECKILL_QUEUE)
                .maxLength(50000) // 最多缓存 5 万条
                .build();
    }
}

@RestController
public class SeckillController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @PostMapping("/seckill")
    public ResponseEntity<String> seckill(@RequestParam String userId, @RequestParam String goodsId) {
        // 1. 基础校验（如登录、参数合法性）
        if (!validate(userId, goodsId)) {
            return ResponseEntity.badRequest().body("Invalid request");
        }
        // 2. 快速入队（毫秒级响应）
        rabbitTemplate.convertAndSend(
            SeckillConfig.SECKILL_QUEUE,
            new SeckillRequest(userId, goodsId, System.currentTimeMillis())
        );
        // 3. 立即返回“请求已接收”，不承诺结果
        return ResponseEntity.ok("Request accepted. Please wait for result.");
    }
}

@Component
public class SeckillConsumer {
    @Autowired
    private GoodsService goodsService;
    // 限制每个消费者实例的并发数
    @RabbitListener(queues = SeckillConfig.SECKILL_QUEUE, concurrency = "1-3")
    public void processSeckill(SeckillRequest request) {
        try {
            boolean success = goodsService.trySeckill(request.getUserId(), request.getGoodsId());
            if (success) {
                // 通知用户成功（如 WebSocket / 短信）
                notificationService.notifySuccess(request.getUserId());
            } else {
                notificationService.notifyFailure(request.getUserId());
            }
        } catch (Exception e) {
            log.error("Seckill failed", e);
            // 可记录到死信队列供人工处理
        }
    }
}

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 10  # 每次最多预取 10 条消息
        acknowledge-mode: manual  # 手动 ACK

@RabbitListener(queues = SeckillConfig.SECKILL_QUEUE)
public void processSeckill(SeckillRequest request, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    try {
        boolean success = goodsService.trySeckill(...);
        // 业务处理...
        channel.basicAck(tag, false); // 手动 ACK
    } catch (Exception e) {
        try {
            channel.basicNack(tag, false, true); // 重回队列 or 进入死信
        } catch (IOException ioEx) {
            log.error("Nack failed", ioEx);
        }
    }
}

// 配置
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    if (ack) {
        log.info("Message confirmed");
    } else {
        log.error("Message lost: {}", cause);
        // 可重发或记录 DB
    }
});
// 发送时指定 CorrelationData
rabbitTemplate.convertAndSend(exchange, routingKey, message, 
    msg -> {
        msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        return msg;
    },
    new CorrelationData(UUID.randomUUID().toString())
);

@Bean
public Queue durableQueue() {
    return QueueBuilder.durable("my.queue").build(); // durable=true
}
// 发送时
MessageProperties props = new MessageProperties();
props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);

@RabbitListener(queues = "my.queue")
public void handleMessage(Message message, Channel channel) throws IOException {
    try {
        // 处理业务
        process(message);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    } catch (Exception e) {
        // 根据策略决定是否 requeue
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
    }
}

@Bean
public Queue mainQueue() {
    return QueueBuilder.durable("main.queue")
            .withArgument("x-dead-letter-exchange", "dlx.exchange")
            .withArgument("x-dead-letter-routing-key", "dlq.key")
            .withArgument("x-message-ttl", 10000) // 10秒后进 DLQ
            .build();
}
@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable("dead.letter.queue").build();
}
@Bean
public DirectExchange deadLetterExchange() {
    return new DirectExchange("dlx.exchange");
}
@Bean
public Binding dlqBinding() {
    return BindingBuilder.bind(deadLetterQueue())
            .to(deadLetterExchange())
            .with("dlq.key");
}