<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. 保存訂單到數(shù)據(jù)庫
        orderRepository.save(order);
        // 2. 發(fā)布事件（異步）
        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 {
            // 發(fā)送郵件邏輯
            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;
    // 構(gòu)造函數(shù)、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. 發(fā)布事件（無 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() {
        // 設(shè)置隊列長度限制，防止內(nèi)存溢出
        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. 基礎(chǔ)校驗（如登錄、參數(shù)合法性）
        if (!validate(userId, goodsId)) {
            return ResponseEntity.badRequest().body("Invalid request");
        }
        // 2. 快速入隊（毫秒級響應(yīng)）
        rabbitTemplate.convertAndSend(
            SeckillConfig.SECKILL_QUEUE,
            new SeckillRequest(userId, goodsId, System.currentTimeMillis())
        );
        // 3. 立即返回“請求已接收”，不承諾結(jié)果
        return ResponseEntity.ok("Request accepted. Please wait for result.");
    }
}

@Component
public class SeckillConsumer {
    @Autowired
    private GoodsService goodsService;
    // 限制每個消費者實例的并發(fā)數(shù)
    @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  # 每次最多預(yù)取 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(...);
        // 業(yè)務(wù)處理...
        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);
        // 可重發(fā)或記錄 DB
    }
});
// 發(fā)送時指定 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
}
// 發(fā)送時
MessageProperties props = new MessageProperties();
props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);

@RabbitListener(queues = "my.queue")
public void handleMessage(Message message, Channel channel) throws IOException {
    try {
        // 處理業(yè)務(wù)
        process(message);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    } catch (Exception e) {
        // 根據(jù)策略決定是否 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");
}