#消息队列在微服务中的应用

// Saga 编排器
public class OrderSagaOrchestrator {
    
    @Autowired
    private OrderService orderService;
    @Autowired
    private PaymentService paymentService;
    @Autowired
    private InventoryService inventoryService;
    
    public void executeOrderSaga(OrderRequest request) {
        try {
            // Step 1: 创建订单
            Order order = orderService.createOrder(request);
            
            // Step 2: 支付
            PaymentResult payment = paymentService.pay(order);
            
            // Step 3: 扣减库存
            inventoryService.reserve(order);
            
            // Step 4: 完成订单
            orderService.confirm(order);
            
        } catch (PaymentException e) {
            // 补偿：取消订单
            orderService.cancel(order);
        } catch (InventoryException e) {
            // 补偿：退款
            paymentService.refund(order.getId());
            orderService.cancel(order);
        }
    }
}

// Debezium 配置
public class DebeziumConfig {
    
    @Bean
    public io.debezium.config.Configuration debeziumConfig() {
        return io.debezium.config.Configuration.create()
            .with("name", "mysql-connector")
            .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
            .with("database.hostname", "mysql")
            .with("database.port", 3306)
            .with("database.user", "debezium")
            .with("database.password", "password")
            .with("database.server.id", 184054)
            .with("database.include.list", "mydb")
            .with("table.include.list", "mydb.orders")
            .with("topic.prefix", "dbz")
            .build();
    }
}

@KafkaListener(topics = "order-events")
public void handleOrderEvent(ConsumerRecord<String, OrderEvent> record) {
    OrderEvent event = record.value();
    
    // 业务层幂等处理
    if (orderService.isProcessed(event.getId())) {
        return;
    }
    
    orderService.processEvent(event);
}

// 订单状态机
public enum OrderStatus {
    CREATED,       // 订单创建
    PAID,          // 已支付
    INVENTORY_RESERVED,  // 库存已预留
    SHIPPED,       // 已发货
    COMPLETED,     // 已完成
    CANCELLED      // 已取消
}

// 合法的状态流转
public boolean canTransition(OrderStatus from, OrderStatus to) {
    return switch (from) {
        case CREATED -> to == PAID || to == CANCELLED;
        case PAID -> to == INVENTORY_RESERVED || to == CANCELLED;
        case INVENTORY_RESERVED -> to == SHIPPED || to == CANCELLED;
        case SHIPPED -> to == COMPLETED;
        default -> false;
    };
}

@Scheduled(fixedRate = 60000)
public void checkTimeoutOrders() {
    List<Order> timeoutOrders = orderService.findTimeoutOrders();
    
    for (Order order : timeoutOrders) {
        // 发送超时补偿事件
        eventPublisher.publish(new OrderTimeoutEvent(order.getId()));
    }
}