Spring Cloud Signal实战 DTD技术如何解决分布式系统数据一致性与高并发挑战

引言：分布式系统的核心挑战

在现代微服务架构中，分布式系统面临着两个核心挑战：数据一致性和高并发处理。传统的单体应用可以通过数据库事务保证数据一致性，但在分布式环境下，跨服务的数据操作变得异常复杂。同时，随着用户量的增长，高并发场景下的数据冲突和性能瓶颈成为系统稳定性的关键制约因素。

Spring Cloud Signal（通常指基于Spring Cloud的信号量/事件驱动架构）结合DTD（Distributed Transaction Design，分布式事务设计）技术，为这些问题提供了优雅的解决方案。本文将深入探讨如何利用这些技术构建高可用、高一致性的分布式系统。

一、分布式数据一致性问题的本质

1.1 CAP理论与BASE原则

分布式系统必须在一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance）之间做出权衡。CAP理论指出，三者不可兼得，通常我们选择CP或AP架构。

BASE原则（Basically Available, Soft state, Eventually consistent）为分布式一致性提供了实践指导：

Basically Available：基本可用
Soft state：软状态/柔性状态
Eventually consistent：最终一致性

1.2 典型分布式一致性场景

场景1：电商下单流程

用户下单 → 库存服务扣减库存 → 订单服务创建订单 → 支付服务处理支付 → 物流服务准备发货

任何一个环节失败都可能导致数据不一致：

库存扣减成功但订单创建失败
订单创建成功但支付失败
支付成功但物流服务不可用

场景2：跨银行转账

银行A扣款 → 银行B加款

必须保证要么都成功，要么都失败，不能出现一方扣款另一方未加款的情况。

二、DTD技术架构深度解析

2.1 DTD核心概念

DTD（Distributed Transaction Design） 不是一个单一技术，而是一套分布式事务设计模式，包含：

TCC（Try-Confirm-Cancel）模式
Saga模式
可靠事件模式
补偿机制

2.2 TCC模式详解

TCC将事务分解为三个阶段：

// TCC接口定义 public interface TccService { // Try阶段：资源预留 boolean tryPhase(OrderDTO order); // Confirm阶段：确认提交 boolean confirmPhase(String transactionId); // Cancel阶段：回滚释放资源 boolean cancelPhase(String transactionId); } // 库存服务的TCC实现 @Service public class InventoryTccService implements TccService { @Autowired private InventoryMapper inventoryMapper; @Autowired private TransactionLogMapper transactionLogMapper; @Override @Transactional(rollbackFor = Exception.class) public boolean tryPhase(OrderDTO order) { // 1. 检查库存 Inventory inventory = inventoryMapper.selectBySku(order.getSkuId()); if (inventory.getStock() < order.getQuantity()) { throw new BusinessException("库存不足"); } // 2. 预扣库存（冻结库存） int updateCount = inventoryMapper.freezeStock( order.getSkuId(), order.getQuantity() ); if (updateCount == 0) { throw new BusinessException("库存预扣失败"); } // 3. 记录事务日志 TransactionLog log = new TransactionLog(); log.setTransactionId(order.getTransactionId()); log.setServiceName("inventory-service"); log.setMethod("tryPhase"); log.setStatus("SUCCESS"); log.setCreateTime(new Date()); transactionLogMapper.insert(log); return true; } @Override @Transactional(rollbackFor = Exception.class) public boolean confirmPhase(String transactionId) { // 1. 查询事务日志 TransactionLog log = transactionLogMapper.selectByTransactionId(transactionId); if (log == null || !"SUCCESS".equals(log.getStatus())) { return false; } // 2. 确认扣减（将冻结库存转为实际扣减） int updateCount = inventoryMapper.confirmStock( log.getSkuId(), log.getQuantity() ); // 3. 删除事务日志（可选） transactionLogMapper.deleteByTransactionId(transactionId); return updateCount > 0; } @Override @Transactional(rollbackFor = Exception.class) public boolean cancelPhase(String transactionId) { // 1. 查询事务日志 TransactionLog log = transactionLogMapper.selectByTransactionId(transactionId); if (log == null) { return true; // 已经取消过 } // 2. 解冻库存 int updateCount = inventoryMapper.unfreezeStock( log.getSkuId(), log.getQuantity() ); // 3. 删除事务日志 transactionLogMapper.deleteByTransactionId(transactionId); return updateCount > 0; } }

2.3 Saga模式详解

Saga模式通过一系列本地事务和补偿事务来实现分布式事务：

// Saga协调器 @Component public class OrderSagaCoordinator { @Autowired private OrderService orderService; @Autowired private InventoryClient inventoryClient; @Autowired private PaymentClient paymentClient; @Autowired private SagaStateRepository sagaStateRepository; /** * 创建订单的Saga流程 */ @Transactional public void createOrderSaga(OrderDTO order) { String sagaId = UUID.randomUUID().toString(); try { // Step 1: 创建订单（本地事务） sagaStateRepository.saveState(sagaId, "ORDER_CREATED"); Order orderEntity = orderService.createOrder(order); // Step 2: 扣减库存（远程调用） sagaStateRepository.saveState(sagaId, "INVENTORY_DEDUCTING"); boolean inventoryResult = inventoryClient.deduct(order); if (!inventoryResult) { throw new BusinessException("库存扣减失败"); } // Step 3: 支付处理（远程调用） sagaStateRepository.saveState(sagaId, "PAYMENT_PROCESSING"); boolean paymentResult = paymentClient.processPayment(order); if (!paymentResult) { // 触发补偿 compensate(sagaId, "PAYMENT_FAILED"); return; } // Saga成功完成 sagaStateRepository.saveState(sagaId, "COMPLETED"); } catch (Exception e) { // 异常时触发补偿 compensate(sagaId, e.getMessage()); } } /** * 补偿方法 */ private void compensate(String sagaId, String failureReason) { SagaState state = sagaStateRepository.findById(sagaId); if (state == null) { return; } // 根据当前状态执行反向操作 switch (state.getCurrentState()) { case "PAYMENT_PROCESSING": // 支付失败，需要回滚订单和库存 orderService.cancelOrder(state.getOrderId()); inventoryClient.refund(state.getOrderId()); break; case "INVENTORY_DEDUCTING": // 库存扣减失败，回滚订单 orderService.cancelOrder(state.getOrderId()); break; case "ORDER_CREATED": // 订单创建失败，无需补偿 break; } sagaStateRepository.saveState(sagaId, "COMPENSATED", failureReason); } } // Saga状态实体 @Entity public class SagaState { @Id private String sagaId; private String currentState; private String orderId; private String failureReason; private Date createTime; private Date updateTime; // getters and setters }

2.4 可靠事件模式

基于消息队列的最终一致性方案：

// 事件生产者 @Service public class OrderEventProducer { @Autowired private KafkaTemplate<String, String> kafkaTemplate; @Autowired private OrderEventRepository eventRepository; /** * 发布订单创建事件 */ @Transactional public void publishOrderCreatedEvent(Order order) { // 1. 保存事件到本地数据库（确保消息不丢失） OrderEvent event = new OrderEvent(); event.setEventId(UUID.randomUUID().toString()); event.setOrderId(order.getId()); event.setEventType("ORDER_CREATED"); event.setPayload(JSON.toJSONString(order)); event.setStatus("PENDING"); event.setCreateTime(new Date()); eventRepository.save(event); // 2. 发送消息到Kafka try { kafkaTemplate.send("order-events", event.getEventId(), event.getPayload()); // 3. 更新事件状态 event.setStatus("SENT"); eventRepository.save(event); } catch (Exception e) { // 发送失败，事件状态保持PENDING，由定时任务重试 log.error("Failed to send event", e); throw e; } } } // 事件消费者 @Component public class InventoryEventConsumer { @Autowired private InventoryService inventoryService; @KafkaListener(topics = "order-events", groupId = "inventory-group") public void handleOrderCreatedEvent(String message) { try { OrderEvent event = JSON.parseObject(message, OrderEvent.class); // 幂等性检查 if (inventoryService.isEventProcessed(event.getEventId())) { return; } // 扣减库存 inventoryService.deductStock(event.getOrderId(), event.getSkuId(), event.getQuantity()); // 发布库存扣减成功事件 publishInventoryDeductedEvent(event.getOrderId()); } catch (Exception e) { // 消费失败，消息重试或进入死信队列 log.error("Failed to consume event", e); throw e; // 触发Kafka重试机制 } } private void publishInventoryDeductedEvent(String orderId) { // 发布新事件，驱动Saga流程下一步 kafkaTemplate.send("inventory-events", orderId, "INVENTORY_DEDUCTED"); } }

三、Spring Cloud Signal与高并发优化

3.1 Spring Cloud Signal核心组件

Spring Cloud Signal通常指基于Spring Cloud的信号量控制和事件驱动架构：

// 信号量配置 @Configuration public class SignalConfig { @Bean public RateLimiter rateLimiter() { return RateLimiter.create(100.0); // 每秒100个请求 } @Bean public Semaphore concurrencySemaphore() { return new Semaphore(50); // 最大并发数50 } } // 信号量限流服务 @Service public class SignalBasedThrottlingService { @Autowired private RateLimiter rateLimiter; @Autowired private Semaphore concurrencySemaphore; /** * 基于信号量的请求处理 */ public <T> T executeWithThrottling(Supplier<T> operation) { // 1. 速率限制检查 if (!rateLimiter.tryAcquire()) { throw new RateLimitExceededException("请求过于频繁，请稍后重试"); } // 2. 并发数限制检查 boolean acquired = false; try { acquired = concurrencySemaphore.tryAcquire(1, TimeUnit.SECONDS); if (!acquired) { throw new ConcurrencyLimitExceededException("系统繁忙，请稍后重试"); } // 3. 执行业务操作 return operation.get(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new BusinessException("请求被中断"); } finally { if (acquired) { concurrencySemaphore.release(); } } } }

3.2 分布式锁与信号量

// Redis分布式锁实现 @Component public class RedisDistributedLock { @Autowired private StringRedisTemplate redisTemplate; private static final String LOCK_KEY_PREFIX = "lock:"; private static final long DEFAULT_EXPIRE_TIME = 30; // 30秒 /** * 尝试获取分布式锁 */ public boolean tryLock(String resource, String requestId, long expireTime) { String script = "if redis.call('exists', KEYS[1]) == 0 then " + " redis.call('set', KEYS[1], ARGV[1]); " + " redis.call('expire', KEYS[1], ARGV[2]); " + " return 1; " + "elseif redis.call('get', KEYS[1]) == ARGV[1] then " + " return 1; " + "else " + " return 0; " + "end"; return redisTemplate.execute( new DefaultRedisScript<>(script, Boolean.class), Collections.singletonList(LOCK_KEY_PREFIX + resource), requestId, String.valueOf(expireTime) ); } /** * 释放分布式锁 */ public boolean unlock(String resource, String requestId) { String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " + " return redis.call('del', KEYS[1]); " + "else " + " return 0; " + "end"; return redisTemplate.execute( new DefaultRedisScript<>(script, Boolean.class), Collections.singletonList(LOCK_KEY_PREFIX + resource), requestId ); } } // 基于分布式锁的库存扣减 @Service public class LockBasedInventoryService { @Autowired private RedisDistributedLock distributedLock; @Autowired private InventoryMapper inventoryMapper; /** * 高并发下的库存扣减 */ public boolean deductStock(Long skuId, Integer quantity) { String lockKey = "inventory:" + skuId; String requestId = UUID.randomUUID().toString(); // 尝试获取锁 boolean locked = distributedLock.tryLock(lockKey, requestId, 30); if (!locked) { throw new ConcurrencyLimitExceededException("库存操作繁忙，请重试"); } try { // 执行库存扣减 Inventory inventory = inventoryMapper.selectBySku(skuId); if (inventory.getStock() < quantity) { return false; } return inventoryMapper.deductStock(skuId, quantity) > 0; } finally { distributedLock.unlock(lockKey, requestId); } } }

3.3 异步化与事件驱动

// 异步事件总线 @Component public class AsyncEventBus { @Autowired private ThreadPoolExecutor asyncExecutor; private final Map<Class<?>, List<Consumer<Object>>> handlers = new ConcurrentHashMap<>(); /** * 注册事件处理器 */ public <T> void register(Class<T> eventType, Consumer<T> handler) { handlers.computeIfAbsent(eventType, k -> new CopyOnWriteArrayList<>()) .add((Consumer<Object>) handler); } /** * 发布事件（异步） */ public <T> void publish(T event) { Class<?> eventType = event.getClass(); List<Consumer<Object>> eventHandlers = handlers.get(eventType); if (eventHandlers != null) { for (Consumer<Object> handler : eventHandlers) { asyncExecutor.submit(() -> { try { handler.accept(event); } catch (Exception e) { log.error("Event handler failed for event: " + event, e); // 可以在这里添加重试或死信处理 } }); } } } } // 使用示例 @Service public class OrderService { @Autowired private AsyncEventBus eventBus; @Autowired private OrderRepository orderRepository; @Transactional public void createOrder(OrderDTO orderDTO) { // 1. 创建订单（同步） Order order = new Order(); order.setOrderId(orderDTO.getOrderId()); order.setAmount(orderDTO.getAmount()); order.setStatus("CREATED"); orderRepository.save(order); // 2. 发布事件（异步） OrderCreatedEvent event = new OrderCreatedEvent(order.getOrderId(), orderDTO.getItems()); eventBus.publish(event); // 3. 后续流程（库存扣减、支付等）由事件处理器异步执行 } } // 事件处理器 @Component public class OrderCreatedEventHandler { @Autowired private InventoryClient inventoryClient; @PostConstruct public void init() { // 注册事件处理器 eventBus.register(OrderCreatedEvent.class, this::handleOrderCreated); } private void handleOrderCreated(OrderCreatedEvent event) { // 异步扣减库存 inventoryClient.deductStock(event.getOrderId(), event.getItems()); } }

四、综合实战：高并发电商下单系统

4.1 系统架构设计

┌─────────────────────────────────────────────────────────────┐ │ API网关层（限流/认证） │ └─────────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────────┐ │ 订单服务（Saga协调器） │ │ - TCC事务管理 │ │ - 信号量限流 │ │ - 异步事件发布 │ └─────────────────────────────────────────────────────────────┘ │ │ │ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 库存服务 │ │ 支付服务 │ │ 物流服务 │ │ - TCC实现 │ │ - TCC实现 │ │ - Saga实现 │ │ - 分布式锁 │ │ - 并发控制 │ │ - 事件驱动 │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ └─────────────────┴─────────────────┘ │ ┌─────────────┐ │ 消息队列 │ │ (Kafka/RabbitMQ)│ └─────────────┘

4.2 完整代码实现

4.2.1 订单服务（Saga协调器）

@RestController @RequestMapping("/api/orders") public class OrderController { @Autowired private OrderSagaCoordinator sagaCoordinator; @Autowired private SignalBasedThrottlingService throttlingService; @PostMapping public ResponseEntity<OrderResponse> createOrder(@RequestBody OrderDTO orderDTO) { // 1. 信号量限流 return throttlingService.executeWithThrottling(() -> { // 2. 参数校验 validateOrder(orderDTO); // 3. 执行Saga事务 sagaCoordinator.createOrderSaga(orderDTO); return ResponseEntity.ok(new OrderResponse( orderDTO.getOrderId(), "SUCCESS", "订单创建成功" )); }); } private void validateOrder(OrderDTO orderDTO) { if (orderDTO.getItems() == null || orderDTO.getItems().isEmpty()) { throw new BusinessException("订单商品不能为空"); } if (orderDTO.getAmount() == null || orderDTO.getAmount().compareTo(BigDecimal.ZERO) <= 0) { throw new BusinessException("订单金额必须大于0"); } } } // Saga协调器（完整版） @Component public class OrderSagaCoordinator { @Autowired private OrderService orderService; @Autowired private InventoryClient inventoryClient; @Autowired private PaymentClient paymentClient; @Autowired private LogisticsClient logisticsClient; @Autowired private SagaStateRepository sagaStateRepository; @Autowired private KafkaTemplate<String, String> kafkaTemplate; /** * 完整的Saga流程 */ @Transactional public void createOrderSaga(OrderDTO orderDTO) { String sagaId = UUID.randomUUID().toString(); String orderId = orderDTO.getOrderId(); try { // ========== Step 1: 创建订单（本地事务） ========== sagaStateRepository.saveState(sagaId, "ORDER_CREATING", orderId); Order order = orderService.createOrder(orderDTO); sagaStateRepository.saveState(sagaId, "ORDER_CREATED", orderId); // ========== Step 2: 预扣库存（TCC Try阶段） ========== sagaStateRepository.saveState(sagaId, "INVENTORY_TRY", orderId); boolean inventoryTry = inventoryClient.tryDeduct(orderId, orderDTO.getItems()); if (!inventoryTry) { throw new BusinessException("库存预扣失败"); } // ========== Step 3: 支付处理（TCC Try阶段） ========== sagaStateRepository.saveState(sagaId, "PAYMENT_TRY", orderId); boolean paymentTry = paymentClient.tryPay(orderId, orderDTO.getAmount()); if (!paymentTry) { // 回滚库存 inventoryClient.cancelDeduct(orderId); throw new BusinessException("支付预扣失败"); } // ========== Step 4: 预约物流（Saga本地事务） ========== sagaStateRepository.saveState(sagaId, "LOGISTICS_BOOKING", orderId); boolean logisticsBook = logisticsClient.book(orderId, orderDTO.getAddress()); if (!logisticsBook) { // 触发补偿 compensate(sagaId, "LOGISTICS_BOOKING_FAILED"); return; } // ========== Step 5: 确认所有TCC阶段 ========== sagaStateRepository.saveState(sagaId, "CONFIRMING", orderId); // 确认库存 inventoryClient.confirmDeduct(orderId); // 确认支付 paymentClient.confirmPay(orderId); // ========== Step 6: 完成Saga ========== sagaStateRepository.saveState(sagaId, "COMPLETED", orderId); orderService.updateOrderStatus(orderId, "SUCCESS"); // 发布成功事件 publishSagaCompletedEvent(sagaId, orderId); } catch (Exception e) { log.error("Saga执行失败: {}", sagaId, e); compensate(sagaId, e.getMessage()); throw new BusinessException("订单创建失败: " + e.getMessage()); } } /** * 补偿机制 */ private void compensate(String sagaId, String failureReason) { SagaState state = sagaStateRepository.findById(sagaId); if (state == null) { return; } String orderId = state.getOrderId(); String currentState = state.getCurrentState(); log.warn("开始补偿: sagaId={}, currentState={}, reason={}", sagaId, currentState, failureReason); // 根据状态执行补偿 try { switch (currentState) { case "LOGISTICS_BOOKING": // 物流预约失败，回滚支付和库存 paymentClient.cancelPay(orderId); inventoryClient.cancelDeduct(orderId); orderService.updateOrderStatus(orderId, "FAILED"); break; case "PAYMENT_TRY": // 支付失败，回滚库存 inventoryClient.cancelDeduct(orderId); orderService.updateOrderStatus(orderId, "FAILED"); break; case "INVENTORY_TRY": // 库存失败，无需补偿 orderService.updateOrderStatus(orderId, "FAILED"); break; case "ORDER_CREATING": // 订单创建失败，无需补偿 break; default: // 其他状态，尝试全部回滚 try { paymentClient.cancelPay(orderId); } catch (Exception e) { log.error("支付补偿失败", e); } try { inventoryClient.cancelDeduct(orderId); } catch (Exception e) { log.error("库存补偿失败", e); } orderService.updateOrderStatus(orderId, "FAILED"); } sagaStateRepository.saveState(sagaId, "COMPENSATED", failureReason); } catch (Exception e) { log.error("补偿执行失败: {}", sagaId, e); // 记录到死信队列，人工介入 publishCompensationFailedEvent(sagaId, orderId, failureReason, e.getMessage()); } } private void publishSagaCompletedEvent(String sagaId, String orderId) { Map<String, Object> event = new HashMap<>(); event.put("sagaId", sagaId); event.put("orderId", orderId); event.put("timestamp", System.currentTimeMillis()); kafkaTemplate.send("saga-completed", sagaId, JSON.toJSONString(event)); } private void publishCompensationFailedEvent(String sagaId, String orderId, String reason, String error) { Map<String, Object> event = new HashMap<>(); event.put("sagaId", sagaId); event.put("orderId", orderId); event.put("reason", reason); event.put("error", error); event.put("timestamp", System.currentTimeMillis()); kafkaTemplate.send("compensation-failed", sagaId, JSON.toJSONString(event)); } }

4.2.2 库存服务（TCC实现）

@RestController @RequestMapping("/api/inventory") public class InventoryController { @Autowired private InventoryTccService tccService; @Autowired private RedisDistributedLock lock; /** * TCC Try接口 */ @PostMapping("/try-deduct") public boolean tryDeduct(@RequestBody InventoryRequest request) { // 使用分布式锁防止并发冲突 String lockKey = "inventory:try:" + request.getSkuId(); String requestId = UUID.randomUUID().toString(); boolean locked = lock.tryLock(lockKey, requestId, 10); if (!locked) { throw new ConcurrencyLimitExceededException("库存操作繁忙"); } try { return tccService.tryPhase(request); } finally { lock.unlock(lockKey, requestId); } } /** * TCC Confirm接口 */ @PostMapping("/confirm-deduct") public boolean confirmDeduct(@RequestBody String transactionId) { return tccService.confirmPhase(transactionId); } /** * TCC Cancel接口 */ @PostMapping("/cancel-deduct") public boolean cancelDeduct(@RequestBody String transactionId) { return tccService.cancelPhase(transactionId); } } // 库存TCC服务实现（完整版） @Service public class InventoryTccService { @Autowired private InventoryMapper inventoryMapper; @Autowired private TransactionLogMapper transactionLogMapper; @Autowired private KafkaTemplate<String, String> kafkaTemplate; @Transactional(rollbackFor = Exception.class) public boolean tryPhase(InventoryRequest request) { // 1. 检查库存 Inventory inventory = inventoryMapper.selectBySku(request.getSkuId()); if (inventory == null || inventory.getStock() < request.getQuantity()) { throw new BusinessException("库存不足: SKU=" + request.getSkuId()); } // 2. 预扣库存（冻结） int updateCount = inventoryMapper.freezeStock( request.getSkuId(), request.getQuantity() ); if (updateCount == 0) { throw new BusinessException("库存预扣失败"); } // 3. 记录事务日志 TransactionLog log = new TransactionLog(); log.setTransactionId(request.getTransactionId()); log.setServiceName("inventory-service"); log.setMethod("tryPhase"); log.setSkuId(request.getSkuId()); log.setQuantity(request.getQuantity()); log.setStatus("SUCCESS"); log.setCreateTime(new Date()); transactionLogMapper.insert(log); // 4. 发布库存冻结事件 publishInventoryFrozenEvent(request); return true; } @Transactional(rollbackFor = Exception.class) public boolean confirmPhase(String transactionId) { // 1. 查询事务日志 TransactionLog log = transactionLogMapper.selectByTransactionId(transactionId); if (log == null) { return true; // 已经确认过 } // 2. 确认扣减（冻结转实际扣减） int updateCount = inventoryMapper.confirmStock( log.getSkuId(), log.getQuantity() ); // 3. 删除事务日志（或标记为已完成） transactionLogMapper.deleteByTransactionId(transactionId); // 4. 发布库存扣减完成事件 publishInventoryConfirmedEvent(log); return updateCount > 0; } @Transactional(rollbackFor = Exception.class) public boolean cancelPhase(String transactionId) { // 1. 查询事务日志 TransactionLog log = transactionLogMapper.selectByTransactionId(transactionId); if (log == null) { return true; // 已经取消过 } // 2. 解冻库存 int updateCount = inventoryMapper.unfreezeStock( log.getSkuId(), log.getQuantity() ); // 3. 删除事务日志 transactionLogMapper.deleteByTransactionId(transactionId); // 4. 发布库存解冻事件 publishInventoryUnfrozenEvent(log); return updateCount > 0; } private void publishInventoryFrozenEvent(InventoryRequest request) { Map<String, Object> event = new HashMap<>(); event.put("type", "INVENTORY_FROZEN"); event.put("skuId", request.getSkuId()); event.put("quantity", request.getQuantity()); event.put("transactionId", request.getTransactionId()); event.put("timestamp", System.currentTimeMillis()); kafkaTemplate.send("inventory-events", request.getTransactionId(), JSON.toJSONString(event)); } private void publishInventoryConfirmedEvent(TransactionLog log) { Map<String, Object> event = new HashMap<>(); event.put("type", "INVENTORY_CONFIRMED"); event.put("skuId", log.getSkuId()); event.put("quantity", log.getQuantity()); event.put("transactionId", log.getTransactionId()); event.put("timestamp", System.currentTimeMillis()); kafkaTemplate.send("inventory-events", log.getTransactionId(), JSON.toJSONString(event)); } private void publishInventoryUnfrozenEvent(TransactionLog log) { Map<String, Object> event = new HashMap<>(); event.put("type", "INVENTORY_UNFROZEN"); event.put("skuId", log.getSkuId()); event.put("quantity", log.getQuantity()); event.put("transactionId", log.getTransactionId()); event.put("timestamp", System.currentTimeMillis()); kafkaTemplate.send("inventory-events", log.getTransactionId(), JSON.toJSONString(event)); } }

4.2.3 支付服务（TCC实现）

@Service public class PaymentTccService { @Autowired private PaymentMapper paymentMapper; @Autowired private AccountMapper accountMapper; @Autowired private TransactionLogMapper transactionLogMapper; @Transactional(rollbackFor = Exception.class) public boolean tryPhase(PaymentRequest request) { // 1. 检查账户余额 Account account = accountMapper.selectByUserId(request.getUserId()); if (account.getBalance().compareTo(request.getAmount()) < 0) { throw new BusinessException("账户余额不足"); } // 2. 冻结资金 int updateCount = accountMapper.freezeBalance( request.getUserId(), request.getAmount() ); if (updateCount == 0) { throw new BusinessException("资金冻结失败"); } // 3. 记录支付流水 Payment payment = new Payment(); payment.setTransactionId(request.getTransactionId()); payment.setUserId(request.getUserId()); payment.setAmount(request.getAmount()); payment.setStatus("FROZEN"); paymentMapper.insert(payment); // 4. 记录事务日志 TransactionLog log = new TransactionLog(); log.setTransactionId(request.getTransactionId()); log.setServiceName("payment-service"); log.setMethod("tryPhase"); log.setAmount(request.getAmount()); log.setStatus("SUCCESS"); log.setCreateTime(new Date()); transactionLogMapper.insert(log); return true; } @Transactional(rollbackFor = Exception.class) public boolean confirmPhase(String transactionId) { // 1. 查询事务日志 TransactionLog log = transactionLogMapper.selectByTransactionId(transactionId); if (log == null) { return true; } // 2. 确认扣款（冻结转实际扣减） Payment payment = paymentMapper.selectByTransactionId(transactionId); int updateCount = accountMapper.confirmDeduct( payment.getUserId(), payment.getAmount() ); if (updateCount > 0) { // 3. 更新支付状态 paymentMapper.updateStatus(transactionId, "SUCCESS"); // 4. 删除事务日志 transactionLogMapper.deleteByTransactionId(transactionId); } return updateCount > 0; } @Transactional(rollbackFor = Exception.class) public boolean cancelPhase(String transactionId) { // 1. 查询事务日志 TransactionLog log = transactionLogMapper.selectByTransactionId(transactionId); if (log == null) { return true; } // 2. 解冻资金 Payment payment = paymentMapper.selectByTransactionId(transactionId); int updateCount = accountMapper.unfreezeBalance( payment.getUserId(), payment.getAmount() ); // 3. 更新支付状态 paymentMapper.updateStatus(transactionId, "CANCELLED"); // 4. 删除事务日志 transactionLogMapper.deleteByTransactionId(transactionId); return updateCount > 0; } }

4.2.4 高并发优化配置

@Configuration public class HighConcurrencyConfig { /** * Tomcat线程池优化 */ @Bean public TomcatServletWebServerFactory servletContainer() { TomcatServletWebServerFactory factory = new TomcatServletWebServerFactory(); factory.addConnectorCustomizers(connector -> { Http11NioProtocol protocol = (Http11NioProtocol) connector.getProtocolHandler(); // 最大线程数 protocol.setMaxThreads(200); // 最小空闲线程数 protocol.setMinSpareThreads(20); // 最大连接数 protocol.setMaxConnections(10000); // 接收队列大小 protocol.setAcceptCount(100); }); return factory; } /** * 异步线程池 */ @Bean public ThreadPoolExecutor asyncExecutor() { return new ThreadPoolExecutor( 20, // 核心线程数 100, // 最大线程数 60L, // 空闲线程存活时间 TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000), // 队列容量 new ThreadFactory() { private final AtomicInteger count = new AtomicInteger(1); @Override public Thread newThread(Runnable r) { Thread t = new Thread(r, "async-executor-" + count.getAndIncrement()); t.setDaemon(false); return t; } }, new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略 ); } /** * Hystrix熔断配置 */ @Bean public HystrixCommand.Setter hystrixCommandSetter() { return HystrixCommand.Setter.withGroupKey( HystrixCommandGroupKey.Factory.asKey("OrderService") ).andCommandPropertiesDefaults( HystrixCommandProperties.Setter() // 超时时间 .withExecutionTimeoutInMilliseconds(5000) // 滑动窗口大小 .withCircuitBreakerRequestVolumeThreshold(20) // 错误阈值 .withCircuitBreakerErrorThresholdPercentage(50) // 休眠时间 .withCircuitBreakerSleepWindowInMilliseconds(10000) ); } }

4.3 监控与告警

// 监控指标收集 @Component public class SagaMetricsCollector { @Autowired private MeterRegistry meterRegistry; private Counter sagaSuccessCounter; private Counter sagaFailureCounter; private Counter compensationCounter; private Timer sagaDurationTimer; @PostConstruct public void init() { sagaSuccessCounter = Counter.builder("saga.success") .description("成功完成的Saga数量") .register(meterRegistry); sagaFailureCounter = Counter.builder("saga.failure") .description("失败的Saga数量") .register(meterRegistry); compensationCounter = Counter.builder("saga.compensation") .description("触发补偿的Saga数量") .register(meterRegistry); sagaDurationTimer = Timer.builder("saga.duration") .description("Saga执行时长") .register(meterRegistry); } public void recordSagaSuccess(String sagaId, long duration) { sagaSuccessCounter.increment(); sagaDurationTimer.record(duration, TimeUnit.MILLISECONDS); } public void recordSagaFailure(String sagaId, String error) { sagaFailureCounter.increment(); // 可以发送告警 sendAlert("Saga失败: " + sagaId + ", 错误: " + error); } public void recordCompensation(String sagaId, String reason) { compensationCounter.increment(); // 补偿次数过多告警 if (compensationCounter.count() > 100) { sendAlert("补偿次数过多，请检查系统稳定性"); } } private void sendAlert(String message) { // 实现告警逻辑，如发送邮件、短信、钉钉通知等 log.error("ALERT: {}", message); } }

五、最佳实践与性能优化

5.1 事务粒度控制

原则：事务范围尽可能小，减少锁持有时间。

// ❌ 错误：大事务 @Transactional public void wrongMethod() { // 耗时操作 remoteService.call(); // 数据库操作 repository.save(); // 另一个耗时操作 anotherRemoteService.call(); } // ✅ 正确：拆分事务 public void correctMethod() { // 耗时操作放在事务外 remoteService.call(); // 小事务 transactionTemplate.execute(status -> { repository.save(); return null; }); anotherRemoteService.call(); }

5.2 幂等性设计

// 幂等性过滤器 @Component public class IdempotencyFilter implements Filter { @Autowired private StringRedisTemplate redisTemplate; private static final String IDEMPOTENCY_KEY_PREFIX = "idempotency:"; @Override public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException { HttpServletRequest httpRequest = (HttpServletRequest) request; String requestId = httpRequest.getHeader("X-Request-ID"); if (requestId == null) { chain.doFilter(request, response); return; } String key = IDEMPOTENCY_KEY_PREFIX + requestId; // 检查是否已处理 String processed = redisTemplate.opsForValue().get(key); if ("PROCESSED".equals(processed)) { // 已处理，直接返回之前的结果 ((HttpServletResponse) response).setStatus(HttpServletResponse.SC_OK); response.getWriter().write("{"code":0,"message":"请求已处理"}"); return; } // 标记为处理中 redisTemplate.opsForValue().set(key, "PROCESSING", 5, TimeUnit.MINUTES); try { chain.doFilter(request, response); // 处理成功，标记为已处理 redisTemplate.opsForValue().set(key, "PROCESSED", 5, TimeUnit.MINUTES); } catch (Exception e) { // 处理失败，删除标记 redisTemplate.delete(key); throw e; } } }

5.3 批量处理优化

// 批量TCC处理 @Service public class BatchInventoryService { @Autowired private InventoryMapper inventoryMapper; /** * 批量预扣库存（减少数据库交互次数） */ @Transactional(rollbackFor = Exception.class) public Map<String, Boolean> batchTryDeduct(List<InventoryRequest> requests) { Map<String, Boolean> results = new HashMap<>(); // 1. 批量检查库存 List<Long> skuIds = requests.stream() .map(InventoryRequest::getSkuId) .collect(Collectors.toList()); List<Inventory> inventories = inventoryMapper.selectBySkuIds(skuIds); Map<Long, Inventory> inventoryMap = inventories.stream() .collect(Collectors.toMap(Inventory::getSkuId, i -> i)); // 2. 批量预扣 List<InventoryRequest> validRequests = new ArrayList<>(); for (InventoryRequest request : requests) { Inventory inventory = inventoryMap.get(request.getSkuId()); if (inventory != null && inventory.getStock() >= request.getQuantity()) { validRequests.add(request); results.put(request.getTransactionId(), true); } else { results.put(request.getTransactionId(), false); } } // 3. 执行批量更新 if (!validRequests.isEmpty()) { inventoryMapper.batchFreezeStock(validRequests); // 4. 批量记录日志 List<TransactionLog> logs = validRequests.stream().map(req -> { TransactionLog log = new TransactionLog(); log.setTransactionId(req.getTransactionId()); log.setServiceName("inventory-service"); log.setMethod("tryPhase"); log.setSkuId(req.getSkuId()); log.setQuantity(req.getQuantity()); log.setStatus("SUCCESS"); log.setCreateTime(new Date()); return log; }).collect(Collectors.toList()); transactionLogMapper.batchInsert(logs); } return results; } }

5.4 数据库优化

-- 事务日志表（用于TCC） CREATE TABLE transaction_log ( id BIGINT PRIMARY KEY AUTO_INCREMENT, transaction_id VARCHAR(64) NOT NULL UNIQUE, service_name VARCHAR(100) NOT NULL, method VARCHAR(50) NOT NULL, sku_id BIGINT, quantity INT, amount DECIMAL(10,2), status VARCHAR(20) NOT NULL, create_time DATETIME NOT NULL, update_time DATETIME, INDEX idx_transaction_id (transaction_id), INDEX idx_create_time (create_time) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 库存表（支持预扣） CREATE TABLE inventory ( id BIGINT PRIMARY KEY AUTO_INCREMENT, sku_id BIGINT NOT NULL UNIQUE, stock INT NOT NULL DEFAULT 0, frozen_stock INT NOT NULL DEFAULT 0, version INT NOT NULL DEFAULT 0, create_time DATETIME NOT NULL, update_time DATETIME NOT NULL, INDEX idx_sku_id (sku_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- Saga状态表 CREATE TABLE saga_state ( saga_id VARCHAR(64) PRIMARY KEY, current_state VARCHAR(50) NOT NULL, order_id VARCHAR(64), failure_reason TEXT, create_time DATETIME NOT NULL, update_time DATETIME NOT NULL, INDEX idx_order_id (order_id), INDEX idx_create_time (create_time) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

六、常见问题与解决方案

6.1 悬挂问题（Hanging Transaction）

问题：Try阶段成功，但Cancel阶段由于网络问题未执行，导致资源永久冻结。

解决方案：

// 增加悬挂检测 @Transactional(rollbackFor = Exception.class) public boolean cancelPhase(String transactionId) { // 检查是否已经Confirm或Cancel过 TransactionLog log = transactionLogMapper.selectByTransactionId(transactionId); if (log == null) { // 可能是悬挂，检查是否有对应的Confirm记录 if (transactionLogMapper.selectConfirmedLog(transactionId) != null) { log.warn("悬挂事务检测到: {}", transactionId); return true; // 已经Confirm过，无需Cancel } // 真正的悬挂，记录告警 log.error("检测到悬挂事务: {}", transactionId); return false; } // 正常Cancel逻辑... return true; }

6.2 空回滚（Empty Rollback）

问题：Try阶段未执行，但Cancel阶段被调用。

解决方案：

@Transactional(rollbackFor = Exception.class) public boolean cancelPhase(String transactionId) { // 检查Try阶段是否执行 TransactionLog log = transactionLogMapper.selectByTransactionId(transactionId); if (log == null) { // 空回滚，直接返回成功 log.warn("空回滚: {}", transactionId); return true; } // 正常Cancel逻辑... return true; }

6.3 幂等性问题

问题：网络重试导致重复调用。

解决方案：

// 在TCC各阶段都做幂等性检查 @Transactional(rollbackFor = Exception.class) public boolean tryPhase(InventoryRequest request) { // 幂等性检查 TransactionLog log = transactionLogMapper.selectByTransactionId( request.getTransactionId() ); if (log != null && "SUCCESS".equals(log.getStatus())) { log.warn("Try阶段幂等: {}", request.getTransactionId()); return true; } // 正常Try逻辑... } @Transactional(rollbackFor = Exception.class) public boolean confirmPhase(String transactionId) { // 幂等性检查 TransactionLog log = transactionLogMapper.selectByTransactionId(transactionId); if (log == null) { log.warn("Confirm阶段幂等: {}", transactionId); return true; // 已经Confirm过 } // 正常Confirm逻辑... }