微服务架构中数据一致性保证机制深度解析

// 库存服务TCC接口 public interface InventoryTCC { 
   // Try阶段：锁定库存    @TwoPhaseBusinessAction(name = "deductInventory", commitMethod = "confirm", rollbackMethod = "cancel")    void deduct(@BusinessActionContextParameter(paramName = "productId") Long productId, @BusinessActionContextParameter(paramName = "quantity") Integer quantity); 
   // Confirm阶段：确认扣减    void confirm(BusinessActionContext context); 
   // Cancel阶段：取消扣减（释放库存）    void cancel(BusinessActionContext context); 
} 

@Service public class OrderSagaCoordinator { 
   @Autowired    private OrderService orderService; 
   @Autowired    private InventoryService inventoryService; 
   @Autowired    private PaymentService paymentService; 
   public void executeSaga(OrderDTO order) { 
       // 创建订单（T1）        Long orderId = orderService.createOrder(order); 
       try {            // 扣减库存（T2）            inventoryService.deduct(order.getProductId(), order.getQuantity()); 
           // 支付处理（T3）            paymentService.pay(orderId, order.getAmount()); 
       } catch (Exception e) {            // 执行补偿事务            if (/* 支付已执行 */) {                paymentService.refund(orderId); // C3            }            if (/* 库存已扣减 */) {                inventoryService.refund(order.getProductId(), order.getQuantity()); // C2            }            orderService.cancelOrder(orderId); // C1        }    } } 

// 订单服务本地事务 @Transactional public void createOrder(Order order) { 
   // 1. 创建订单（本地事务）    orderMapper.insert(order); 
   // 2. 写入本地消息表（与订单事务同享事务）    Message message = new Message("inventory.deduct", order.getId(), order.getProductId(), order.getQuantity()); 
   messageMapper.insert(message); } 
// 消息发送器（定时任务） @Scheduled(fixedRate = 1000) public void sendPendingMessages() { 
   List<Message> pending = messageMapper.findByStatus(UNSENT);    for (Message msg : pending) {        try {            kafkaTemplate.send(msg.getTopic(), msg.getContent());            messageMapper.updateStatus(msg.getId(), SENT);        } catch (Exception e) {            // 重试次数超限后标记为失败，人工干预            if (msg.getRetryCount() > 3) {                messageMapper.updateStatus(msg.getId(), FAILED);            } else {                messageMapper.incrementRetryCount(msg.getId());            }        }    } } 

@Service public class OrderTransactionMessageService { 
   @Autowired    private RocketMQTemplate rocketMQTemplate; 
   @Autowired    private OrderMapper orderMapper;       public void createOrderWithTransaction(Order order) { 
       // 1. 发送半事务消息        rocketMQTemplate.sendMessageInTransaction(            "order-topic",            MessageBuilder.withPayload(order).build(),            order // 传递到本地事务执行器的参数        );    }    // 2. 本地事务执行器    @RocketMQTransactionListener    class OrderTransactionListener implements RocketMQLocalTransactionListener {        @Override        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) { 
           Order order = (Order) arg;            try {                orderMapper.insert(order); // 执行本地事务                return RocketMQLocalTransactionState.COMMIT; // 提交消息            } catch (Exception e) {                return RocketMQLocalTransactionState.ROLLBACK; // 回滚消息            }        } 
       @Override        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { 
           // 3. 消息回查：检查本地事务状态（如订单是否存在）            String orderId = msg.getHeaders().get("orderId", String.class); 
           return orderMapper.exists(orderId) ? COMMIT : ROLLBACK;        }    } } 

@Service public class InventoryService { 
   @Autowired    private StringRedisTemplate redisTemplate; 
   public void deduct(Long productId, Integer quantity) { 
       String lockKey = "lock:inventory:" + productId; 
       String lockValue = UUID.randomUUID().toString(); 
       try {            // 获取锁（30秒自动释放）            boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 30, TimeUnit.SECONDS); 
           if (!locked) {                throw new RuntimeException("获取锁失败，并发冲突");            }            // 扣减库存业务逻辑            Inventory inventory = inventoryMapper.selectById(productId); 
           if (inventory.getQuantity() < quantity) {                throw new RuntimeException("库存不足");            }            inventoryMapper.deduct(productId, quantity);        } finally {            // 释放锁（判断是否为当前锁，避免误删）            if (lockValue.equals(redisTemplate.opsForValue().get(lockKey))) {                redisTemplate.delete(lockKey);            }        }    } } 

@Scheduled(cron = "0 0 */1 * * ?") // 每小时执行 public void checkAndFixInventoryConsistency() { 
   // 1. 对比订单表已扣减库存与库存表实际库存    List<InventoryMismatch> mismatches = inventoryChecker.findMismatches(); 
   // 2. 修复不一致（如库存少扣则补扣，多扣则回滚）    for (InventoryMismatch mismatch : mismatches) {        if (mismatch.getActualInventory() > mismatch.getExpectedInventory()) { 
           inventoryService.deduct(mismatch.getProductId(), mismatch.getDiff());        } else { 
           inventoryService.refund(mismatch.getProductId(), -mismatch.getDiff()); 
       }    } } 

@GetMapping("/deduct") public Result deduct(@RequestParam Long productId, @RequestParam Integer quantity, @RequestHeader("Idempotency-Key") String idempotencyKey) { 
   if (redisTemplate.opsForValue().setIfAbsent(idempotencyKey, "1", 1, TimeUnit.HOURS)) { 
       // 执行扣减逻辑        return inventoryService.deduct(productId, quantity);    } else { 
       // 重复请求，返回上次结果        return Result.success("重复请求，已处理");    } } 

UPDATE inventory SET quantity = quantity - 1, version = version + 1  WHERE product_id = ? AND version = ?