// 订单服务 - 强耦合
@Service
public class OrderService {
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private LogisticsService logisticsService;
    
    @Autowired
    private NotificationService notificationService;
    
    public Order createOrder(CreateOrderRequest request) {
        Order order = orderRepository.save(request.toOrder());
        
        // 同步调用，任何一个失败都会导致订单创建失败
        inventoryService.deduct(order.getItems());
        logisticsService.createDelivery(order);
        notificationService.sendOrderCreated(order);
        
        return order;
    }
}

// 订单服务 - 解耦
@Service
public class OrderService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public Order createOrder(CreateOrderRequest request) {
        Order order = orderRepository.save(request.toOrder());
        
        // 发送消息，立即返回
        OrderCreatedEvent event = new OrderCreatedEvent(order);
        rabbitTemplate.convertAndSend("order.events", event);
        
        return order;
    }
}

// 库存服务 - 独立消费
@Component
public class InventoryListener {
    
    @RabbitListener(queues = "inventory.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        inventoryService.deduct(event.getItems());
    }
}

// 物流服务 - 独立消费
@Component
public class LogisticsListener {
    
    @RabbitListener(queues = "logistics.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        logisticsService.createDelivery(event.getOrder());
    }
}

// 用户注册 - 同步处理（慢）
public User register(RegisterRequest request) {
    User user = userRepository.save(request.toUser());
    
    // 以下操作耗时且非核心
    emailService.sendWelcomeEmail(user);  // 200ms
    smsService.sendVerifySms(user);       // 300ms
    analyticsService.trackUserRegister(user); // 100ms
    couponService.grantWelcomeCoupon(user);   // 150ms
    
    return user; // 总耗时：750ms+
}

// 用户注册 - 异步处理（快）
public User register(RegisterRequest request) {
    User user = userRepository.save(request.toUser());
    
    // 发送消息，立即返回
    eventPublisher.publish(new UserRegisteredEvent(user));
    
    return user; // 总耗时：50ms
}

// 秒杀请求 - 削峰处理
@Service
public class SeckillService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public SeckillResult seckill(SeckillRequest request) {
        // 1. 快速验证
        if (!seckillValidator.validate(request)) {
            return SeckillResult.failed("验证失败");
        }
        
        // 2. 写入消息队列（快速返回）
        rabbitTemplate.convertAndSend("seckill.queue", request);
        
        return SeckillResult.processing("排队中");
    }
}

// 后台消费 - 匀速处理
@Component
public class SeckillConsumer {
    
    @RabbitListener(queues = "seckill.queue")
    public void consume(SeckillRequest request) {
        // 以稳定速率处理（2000 QPS）
        seckillProcessor.process(request);
    }
}

@Configuration
public class DirectExchangeConfig {
    
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("order.direct");
    }
    
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable("order.queue")
            .withArgument("x-message-ttl", 60000) // 消息 TTL 60 秒
            .build();
    }
    
    @Bean
    public Binding orderBinding(Queue orderQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(orderQueue)
            .to(directExchange)
            .with("order.created"); // Routing Key
    }
}

// 生产者
rabbitTemplate.convertAndSend("order.direct", "order.created", message);

// 消费者 - 只接收 order.created 消息
@RabbitListener(queues = "order.queue")
public void handleOrderCreated(Message message) {
    // 处理订单创建
}

@Configuration
public class FanoutExchangeConfig {
    
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("order.fanout");
    }
    
    @Bean
    public Queue inventoryQueue() {
        return QueueBuilder.durable("inventory.queue").build();
    }
    
    @Bean
    public Queue logisticsQueue() {
        return QueueBuilder.durable("logistics.queue").build();
    }
    
    @Bean
    public Queue notificationQueue() {
        return QueueBuilder.durable("notification.queue").build();
    }
    
    @Bean
    public Binding inventoryBinding(Queue inventoryQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(inventoryQueue).to(fanoutExchange);
    }
    
    @Bean
    public Binding logisticsBinding(Queue logisticsQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(logisticsQueue).to(fanoutExchange);
    }
    
    @Bean
    public Binding notificationBinding(Queue notificationQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(notificationQueue).to(fanoutExchange);
    }
}

// 生产者 - 消息广播到所有队列
rabbitTemplate.convertAndSend("order.fanout", "", message);

@Configuration
public class TopicExchangeConfig {
    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("order.topic");
    }
    
    @Bean
    public Queue allOrdersQueue() {
        return QueueBuilder.durable("order.all").build();
    }
    
    @Bean
    public Queue createdOrdersQueue() {
        return QueueBuilder.durable("order.created").build();
    }
    
    @Bean
    public Queue paidOrdersQueue() {
        return QueueBuilder.durable("order.paid").build();
    }
    
    @Bean
    public Binding allBinding(Queue allOrdersQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(allOrdersQueue)
            .to(topicExchange)
            .with("order.*"); // 匹配 order.开头的所有消息
    }
    
    @Bean
    public Binding createdBinding(Queue createdOrdersQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(createdOrdersQueue)
            .to(topicExchange)
            .with("order.created"); // 精确匹配
    }
    
    @Bean
    public Binding paidBinding(Queue paidOrdersQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(paidOrdersQueue)
            .to(topicExchange)
            .with("order.paid.*"); // 匹配 order.paid.开头的消息
    }
}

// 生产者
rabbitTemplate.convertAndSend("order.topic", "order.created.vip", message);
rabbitTemplate.convertAndSend("order.topic", "order.paid.normal", message);

@Configuration
public class RabbitConfig {
    
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin123");
        
        // 开启生产者确认
        factory.setPublisherConfirms(true);
        factory.setPublisherReturns(true);
        
        return factory;
    }
    
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        
        // 确认回调
        rabbitTemplate.setConfirmCallback((correlationData, acknowledged, cause) -> {
            if (acknowledged) {
                log.info("消息发送成功：{}", correlationData);
            } else {
                log.error("消息发送失败：{}", cause);
                // 记录失败，后续重试
            }
        });
        
        // 返回回调
        rabbitTemplate.setReturnsCallback(returnedMessage -> {
            log.error("消息被退回：{}", returnedMessage.getMessage());
            // 处理退回消息
        });
        
        return rabbitTemplate;
    }
}

@Configuration
public class ConsumerConfig {
    
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        
        // 手动 ACK
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        
        // 并发消费者数量
        factory.setConcurrentConsumers(5);
        factory.setMaxConcurrentConsumers(10);
        
        // 预取数量
        factory.setPrefetchCount(10);
        
        return factory;
    }
}

// 消费者 - 手动 ACK
@Component
public class OrderConsumer {
    
    @RabbitListener(queues = "order.queue")
    public void handleMessage(Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        
        try {
            // 处理消息
            OrderCreatedEvent event = parseEvent(message);
            orderService.process(event);
            
            // 手动确认
            channel.basicAck(deliveryTag, false);
            
        } catch (Exception e) {
            log.error("处理消息失败", e);
            
            // 判断是否重试
            Integer retryCount = getRetryCount(message);
            if (retryCount < 3) {
                // 拒绝消息，重新入队
                channel.basicNack(deliveryTag, false, true);
            } else {
                // 发送到死信队列
                channel.basicNack(deliveryTag, false, false);
                sendToDeadLetterQueue(message);
            }
        }
    }
}

@Configuration
public class DeadLetterConfig {
    
    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable("dead.letter.queue").build();
    }
    
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange("dead.letter.exchange");
    }
    
    @Bean
    public Binding deadLetterBinding(Queue deadLetterQueue, DirectExchange deadLetterExchange) {
        return BindingBuilder.bind(deadLetterQueue)
            .to(deadLetterExchange)
            .with("dead.letter");
    }
    
    @Bean
    public Queue orderQueueWithDLQ() {
        return QueueBuilder.durable("order.queue")
            .withArgument("x-dead-letter-exchange", "dead.letter.exchange")
            .withArgument("x-dead-letter-routing-key", "dead.letter")
            .withArgument("x-message-ttl", 60000)
            .build();
    }
}

@Configuration
public class DelayQueueConfig {
    
    // 延迟消息 TTL 队列
    @Bean
    public Queue delayQueue() {
        return QueueBuilder.durable("delay.queue")
            .withArgument("x-message-ttl", 60000) // 60 秒
            .withArgument("x-dead-letter-exchange", "order.exchange")
            .withArgument("x-dead-letter-routing-key", "order.timeout")
            .build();
    }
    
    @Bean
    public DirectExchange delayExchange() {
        return new DirectExchange("delay.exchange");
    }
    
    @Bean
    public Binding delayBinding(Queue delayQueue, DirectExchange delayExchange) {
        return BindingBuilder.bind(delayQueue)
            .to(delayExchange)
            .with("order.delay");
    }
    
    // 超时订单队列（死信队列）
    @Bean
    public Queue timeoutOrderQueue() {
        return QueueBuilder.durable("order.timeout.queue").build();
    }
    
    @Bean
    public Binding timeoutBinding(Queue timeoutOrderQueue, DirectExchange delayExchange) {
        return BindingBuilder.bind(timeoutOrderQueue)
            .to(delayExchange)
            .with("order.timeout");
    }
}

// 发送延迟消息
public void sendDelayOrder(Order order) {
    rabbitTemplate.convertAndSend(
        "delay.exchange",
        "order.delay",
        order,
        message -> {
            // 设置延迟时间（毫秒）
            message.getMessageProperties().setDelay(30 * 60 * 1000); // 30 分钟
            return message;
        }
    );
}

// 消费超时订单
@RabbitListener(queues = "order.timeout.queue")
public void handleTimeoutOrder(Order order) {
    orderService.cancelOrder(order.getId());
}

@Configuration
public class KafkaProducerConfig {
    
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        
        // 基础配置
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        
        // 可靠性配置
        config.put(ProducerConfig.ACKS_CONFIG, "all"); // 所有副本确认
        config.put(ProducerConfig.RETRIES_CONFIG, 3); // 重试次数
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 幂等性
        
        // 性能配置
        config.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); // 批次大小 16KB
        config.put(ProducerConfig.LINGER_MS_CONFIG, 10); // 等待时间 10ms
        config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy"); // 压缩
        
        return new DefaultKafkaProducerFactory<>(config);
    }
    
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        KafkaTemplate<String, String> template = new KafkaTemplate<>(producerFactory());
        
        // 发送回调
        template.setProducerListener(new ProducerListener<String, String>() {
            @Override
            public void onSuccess(ProducerRecord<String, String> record, 
                                 RecordMetadata metadata) {
                log.info("发送成功：topic={}, partition={}, offset={}", 
                    metadata.topic(), metadata.partition(), metadata.offset());
            }
            
            @Override
            public void onError(ProducerRecord<String, String> record, 
                               RecordMetadata metadata, Exception exception) {
                log.error("发送失败", exception);
            }
        });
        
        return template;
    }
}

@Configuration
public class KafkaConsumerConfig {
    
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        
        // 基础配置
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        
        // 消费者组
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-consumer-group");
        
        // Offset 提交
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 手动提交
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // 从最新开始
        
        return new DefaultKafkaConsumerFactory<>(config);
    }
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        
        // 并发消费者
        factory.setConcurrency(3);
        
        return factory;
    }
}

// 消费者 - 手动提交 Offset
@Component
public class OrderConsumer {
    
    @KafkaListener(topics = "order-topic", groupId = "order-consumer-group")
    public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            log.info("收到消息：key={}, value={}", record.key(), record.value());
            
            // 处理消息
            orderService.process(record.value());
            
            // 手动提交 Offset
            ack.acknowledge();
            
        } catch (Exception e) {
            log.error("处理消息失败", e);
            // 不提交 Offset，下次重新消费
            throw e;
        }
    }
}

// 生产者 - 保证分区有序
public void sendOrderMessage(Order order) {
    // 使用订单 ID 作为 key，保证同一订单的消息发送到同一分区
    String key = String.valueOf(order.getId());
    kafkaTemplate.send("order-topic", key, order.toJson());
}

// 消费者 - 单线程消费保证顺序
@Component
public class OrderConsumer {
    
    @KafkaListener(
        topics = "order-topic",
        groupId = "order-consumer-group",
        containerFactory = "singleThreadKafkaListenerContainerFactory"
    )
    public void consume(ConsumerRecord<String, String> record) {
        // 单线程处理，保证分区内顺序
        orderService.process(record.value());
    }
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> singleThreadKafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(1); // 单线程
    return factory;
}

@Configuration
public class KafkaExactlyOnceConfig {
    
    @Bean
    public ProducerFactory<String, String> exactlyOnceProducerFactory() {
        Map<String, Object> config = new HashMap<>();
        
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-producer-1"); // 事务 ID
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 幂等性
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        
        return new DefaultKafkaProducerFactory<>(config);
    }
    
    @Bean
    public KafkaTransactionManager<String, String> kafkaTransactionManager() {
        return new KafkaTransactionManager<>(exactlyOnceProducerFactory());
    }
}

// 事务消息
@Service
public class OrderService {
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    @Transactional // 开启事务
    public void createOrder(Order order) {
        // 1. 保存订单到数据库
        orderRepository.save(order);
        
        // 2. 发送消息（同一事务）
        kafkaTemplate.executeInTransaction(template -> {
            template.send("order-topic", order.toJson());
            template.send("inventory-topic", order.getItems().toJson());
            return null;
        });
    }
}

@Configuration
public class RocketMQConfig {
    
    @Value("${rocketmq.name-server}")
    private String nameServer;
    
    @Bean
    public DefaultMQProducer producer() throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer("order-producer-group");
        producer.setNamesrvAddr(nameServer);
        producer.setInstanceName(UUID.randomUUID().toString());
        
        // 重试配置
        producer.setRetryTimesWhenSendFailed(3);
        producer.setRetryTimesWhenSendAsyncFailed(3);
        
        // 超时配置
        producer.setSendMsgTimeout(3000);
        
        producer.start();
        return producer;
    }
}

// 发送消息
@Service
public class OrderMessageService {
    
    @Autowired
    private DefaultMQProducer producer;
    
    // 同步发送
    public void sendSync(Order order) throws Exception {
        Message msg = new Message(
            "order-topic",
            "order-created",
            order.toJson().getBytes(RemotingHelper.DEFAULT_CHARSET)
        );
        
        SendResult result = producer.send(msg);
        log.info("发送成功：msgId={}", result.getMsgId());
    }
    
    // 异步发送
    public void sendAsync(Order order) throws Exception {
        Message msg = new Message(
            "order-topic",
            "order-created",
            order.toJson().getBytes(RemotingHelper.DEFAULT_CHARSET)
        );
        
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送成功：msgId={}", sendResult.getMsgId());
            }
            
            @Override
            public void onException(Throwable e) {
                log.error("发送失败", e);
            }
        });
    }
    
    // 单向发送（不关心结果）
    public void sendOneway(Order order) throws Exception {
        Message msg = new Message(
            "order-topic",
            "order-created",
            order.toJson().getBytes(RemotingHelper.DEFAULT_CHARSET)
        );
        
        producer.sendOneway(msg);
    }
}

@Configuration
public class RocketMQConsumerConfig {
    
    @Value("${rocketmq.name-server}")
    private String nameServer;
    
    @Bean
    public DefaultMQPushConsumer consumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order-consumer-group");
        consumer.setNamesrvAddr(nameServer);
        consumer.setInstanceName(UUID.randomUUID().toString());
        
        // 消费点位
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        
        // 并发配置
        consumer.setConsumeThreadMin(10);
        consumer.setConsumeThreadMax(20);
        
        // 批量消费
        consumer.setConsumeMessageBatchMaxSize(1);
        
        // 重试配置
        consumer.setMaxReconsumeTimes(3);
        
        consumer.subscribe("order-topic", "*");
        
        // 消息监听
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                try {
                    String body = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                    log.info("收到消息：msgId={}, body={}", msg.getMsgId(), body);
                    
                    // 处理消息
                    orderService.process(body);
                    
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    
                } catch (Exception e) {
                    log.error("处理消息失败", e);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        
        consumer.start();
        return consumer;
    }
}

// 事务消息监听器
@Component
public class OrderTransactionListener implements TransactionListener {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private LocalTransactionTable transactionTable;
    
    // 执行本地事务
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            Order order = parseOrder(msg);
            
            // 1. 执行本地事务
            orderService.createOrder(order);
            
            // 2. 记录事务状态
            transactionTable.record(msg.getMsgId(), TransactionStatus.COMMITTED);
            
            return LocalTransactionState.COMMIT_MESSAGE;
            
        } catch (Exception e) {
            log.error("执行本地事务失败", e);
            transactionTable.record(msg.getMsgId(), TransactionStatus.ROLLBACK);
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    // 事务状态回查
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String msgId = msg.getMsgId();
        TransactionStatus status = transactionTable.getStatus(msgId);
        
        log.info("事务回查：msgId={}, status={}", msgId, status);
        
        switch (status) {
            case COMMITTED:
                return LocalTransactionState.COMMIT_MESSAGE;
            case ROLLBACK:
                return LocalTransactionState.ROLLBACK_MESSAGE;
            default:
                return LocalTransactionState.UNKNOW;
        }
    }
}

// 发送事务消息
@Service
public class OrderTransactionService {
    
    @Autowired
    private TransactionMQProducer producer;
    
    public void sendTransactionMessage(Order order) throws Exception {
        Message msg = new Message(
            "order-topic",
            "order-created",
            order.toJson().getBytes(RemotingHelper.DEFAULT_CHARSET)
        );
        
        // 发送事务消息
        TransactionSendResult result = producer.sendMessageInTransaction(msg, order);
        
        log.info("事务消息发送：msgId={}, state={}", 
            result.getMsgId(), result.getLocalTransactionState());
    }
}