// RabbitMQ生產(chǎn)者確認(rèn)配置
@Bean
public RabbitTemplate rabbitTemplate() {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setConfirmCallback((correlationData, ack, cause) -> {
        if (ack) {
            // 消息成功到達(dá)Broker
            messageStatusService.markConfirmed(correlationData.getId());
        } else {
            // 發(fā)送失敗，觸發(fā)重試
            retryService.scheduleRetry(correlationData.getId());
        }
    });
    return template;
}

// 可靠發(fā)送方法
public void sendReliable(String exchange, String routingKey, Object message) {
    String messageId = generateId();
    // 先落庫(kù)保存發(fā)送狀態(tài)
    messageStatusService.saveSendingStatus(messageId, message);
    
    // 發(fā)送持久化消息
    rabbitTemplate.convertAndSend(exchange, routingKey, message, msg -> {
        msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        msg.getMessageProperties().setMessageId(messageId);
        return msg;
    }, new CorrelationData(messageId));
}

// 持久化隊(duì)列配置
@Bean
public Queue orderQueue() {
    return QueueBuilder.durable("order.queue")  // 隊(duì)列持久化
            .deadLetterExchange("order.dlx")    // 死信交換機(jī)
            .build();
}

// 發(fā)送持久化消息
public void sendPersistentMessage(Object message) {
    rabbitTemplate.convertAndSend("order.exchange", "order.create", message, msg -> {
        msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); // 消息持久化
        return msg;
    });
}

// Kafka持久化配置
@Bean
public ProducerFactory<String, Object> producerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.ACKS_CONFIG, "all"); // 所有副本確認(rèn)
    props.put(ProducerConfig.RETRIES_CONFIG, 3);   // 重試次數(shù)
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 冪等性
    returnnew DefaultKafkaProducerFactory<>(props);
}

// 手動(dòng)確認(rèn)消費(fèi)者
@RabbitListener(queues = "order.queue")
public void handleMessage(Order order, Message message, Channel channel) {
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    
    try {
        // 業(yè)務(wù)處理
        orderService.processOrder(order);
        
        // 手動(dòng)確認(rèn)
        channel.basicAck(deliveryTag, false);
        log.info("消息處理完成: {}", order.getOrderId());
        
    } catch (Exception e) {
        log.error("消息處理失敗: {}", order.getOrderId(), e);
        
        // 處理失敗，重新入隊(duì)
        channel.basicNack(deliveryTag, false, true);
    }
}

// 消費(fèi)者容器配置
@Bean
public SimpleRabbitListenerContainerFactory containerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 手動(dòng)確認(rèn)
    factory.setPrefetchCount(10); // 預(yù)取數(shù)量
    factory.setConcurrentConsumers(3); // 并發(fā)消費(fèi)者
    return factory;
}

// 本地事務(wù)表方案
@Transactional
public void createOrder(Order order) {
    // 1. 保存訂單到數(shù)據(jù)庫(kù)
    orderRepository.save(order);
    
    // 2. 保存消息到本地消息表
    LocalMessage localMessage = new LocalMessage();
    localMessage.setBusinessId(order.getOrderId());
    localMessage.setContent(JSON.toJSONString(order));
    localMessage.setStatus(MessageStatus.PENDING);
    localMessageRepository.save(localMessage);
    
    // 3. 事務(wù)提交，本地業(yè)務(wù)和消息存儲(chǔ)保持一致性
}

// 定時(shí)任務(wù)掃描并發(fā)送消息
@Scheduled(fixedDelay = 5000)
public void sendPendingMessages() {
    List<LocalMessage> pendingMessages = localMessageRepository.findByStatus(MessageStatus.PENDING);
    
    for (LocalMessage message : pendingMessages) {
        try {
            // 發(fā)送消息到MQ
            rabbitTemplate.convertAndSend("order.exchange", "order.create", message.getContent());
            
            // 更新消息狀態(tài)為已發(fā)送
            message.setStatus(MessageStatus.SENT);
            localMessageRepository.save(message);
            
        } catch (Exception e) {
            log.error("發(fā)送消息失敗: {}", message.getId(), e);
        }
    }
}

// RocketMQ事務(wù)消息
public void sendTransactionMessage(Order order) {
    TransactionMQProducer producer = new TransactionMQProducer("order_producer");
    
    // 發(fā)送事務(wù)消息
    Message msg = new Message("order_topic", "create", 
                             JSON.toJSONBytes(order));
    
    TransactionSendResult result = producer.sendMessageInTransaction(msg, null);
    
    if (result.getLocalTransactionState() == LocalTransactionState.COMMIT_MESSAGE) {
        log.info("事務(wù)消息提交成功");
    }
}

// 重試隊(duì)列配置
@Bean
public Queue orderQueue() {
    return QueueBuilder.durable("order.queue")
            .withArgument("x-dead-letter-exchange", "order.dlx") // 死信交換機(jī)
            .withArgument("x-dead-letter-routing-key", "order.dead")
            .withArgument("x-message-ttl", 60000) // 60秒后進(jìn)入死信
            .build();
}

// 死信隊(duì)列配置
@Bean
public Queue orderDeadLetterQueue() {
    return QueueBuilder.durable("order.dead.queue").build();
}

// 消費(fèi)者重試邏輯
@RabbitListener(queues = "order.queue")
public void handleMessageWithRetry(Order order, Message message, Channel channel) {
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    
    try {
        orderService.processOrder(order);
        channel.basicAck(deliveryTag, false);
        
    } catch (TemporaryException e) {
        // 臨時(shí)異常，重新入隊(duì)重試
        channel.basicNack(deliveryTag, false, true);
        
    } catch (PermanentException e) {
        // 永久異常，直接確認(rèn)進(jìn)入死信隊(duì)列
        channel.basicAck(deliveryTag, false);
        log.error("消息進(jìn)入死信隊(duì)列: {}", order.getOrderId(), e);
    }
}

// 死信隊(duì)列消費(fèi)者
@RabbitListener(queues = "order.dead.queue")
public void handleDeadLetterMessage(Order order) {
    log.warn("處理死信消息: {}", order.getOrderId());
    // 發(fā)送告警、記錄日志、人工處理等
    alertService.sendAlert("死信消息告警", order.toString());
}