分布式事务原理及实际业务场景详解

一、分布式事务基础概念

1.1 什么是分布式事务？

分布式事务是指跨多个数据库、服务或系统的操作序列，这些操作作为一个整体，要么全部成功，要么全部失败，保证数据的一致性。

1.2 本地事务 vs 分布式事务

二、分布式事务核心挑战

2.1 经典问题：CAP定理

/** * CAP定理：分布式系统最多只能同时满足以下两个特性 * * C - Consistency（一致性）：所有节点在同一时间的数据完全一致 * A - Availability（可用性）：每个请求都能收到响应（不保证数据最新） * P - Partition tolerance（分区容错性）：系统在节点间通信失败时仍能工作 * * 现实选择： * CP系统：ZooKeeper、Etcd（放弃可用性，保证一致性） * AP系统：Cassandra、Eureka（放弃强一致性，保证可用性） */

2.2 BASE理论

/** * BASE理论：对CAP中一致性和可用性权衡的结果 * * Basically Available（基本可用）：系统出现故障时，允许损失部分可用性 * Soft state（软状态）：允许系统存在中间状态，各节点数据可能存在延迟 * Eventually consistent（最终一致性）：经过一段时间后，所有节点数据最终一致 */

2.3 数据一致性问题

// 1. 脏读 // 事务A读取了事务B未提交的数据 // 2. 不可重复读 // 事务A多次读取同一数据，事务B在期间修改并提交了数据 // 3. 幻读 // 事务A多次查询同一范围，事务B在期间插入新数据 // 4. 丢失更新 // 两个事务同时读取并修改同一数据，后提交的覆盖了先提交的

三、分布式事务解决方案

3.1 两阶段提交（2PC）

/** * 2PC - Two-Phase Commit * 角色：协调者（Coordinator）、参与者（Participant） */ public class TwoPhaseCommit { // 第一阶段：准备阶段（投票阶段） public void preparePhase() { // 1. 协调者向所有参与者发送prepare请求 // 2. 参与者执行事务操作，但不提交，记录undo/redo日志 // 3. 参与者返回投票结果（同意/中止） } // 第二阶段：提交阶段（执行阶段） public void commitPhase() { // 情况1：所有参与者都同意 // 协调者发送commit命令 → 参与者提交事务 → 参与者返回ack // 情况2：任一参与者返回中止或超时 // 协调者发送rollback命令 → 参与者回滚事务 → 参与者返回ack } /** * 2PC优缺点分析： * 优点： * - 强一致性 * - 实现相对简单 * * 缺点： * - 同步阻塞（所有参与者等待协调者） * - 单点故障（协调者宕机导致阻塞） * - 数据不一致（第二阶段部分参与者提交失败） * - 保守策略（任一失败则全部回滚） */ }

3.2 三阶段提交（3PC）

/** * 3PC - Three-Phase Commit * 引入超时机制和预提交阶段，解决2PC的阻塞问题 */ public class ThreePhaseCommit { // 第一阶段：CanCommit（询问阶段） public void canCommitPhase() { // 协调者询问参与者是否可以提交 // 参与者检查自身状态，返回Yes/No } // 第二阶段：PreCommit（预提交阶段） public void preCommitPhase() { // 情况1：所有参与者返回Yes // 协调者发送PreCommit → 参与者执行事务，记录日志 // 情况2：有参与者返回No或超时 // 协调者发送abort → 参与者中断事务 } // 第三阶段：DoCommit（提交阶段） public void doCommitPhase() { // 协调者发送doCommit → 参与者提交事务 // 如果协调者故障，参与者在超时后自动提交 } /** * 3PC vs 2PC改进： * 1. 引入超时机制：协调者和参与者都设置超时 * 2. 减少阻塞：PreCommit后参与者可以安全提交 * 3. 仍存在的问题：网络分区可能导致数据不一致 */ }

3.3 TCC（Try-Confirm-Cancel）

/** * TCC - Try Confirm Cancel * 补偿型事务，适用于需要高一致性且性能要求高的场景 */ @Service public class OrderTccService { // 第一阶段：Try（尝试） @Transactional public void tryPhase(OrderDTO order) { // 1. 检查业务约束（库存、余额等） checkBusiness(order); // 2. 预留必要资源 // 库存服务：冻结库存（status = FROZEN） // 账户服务：冻结余额（balance_frozen） // 订单服务：创建待确认订单（status = TRY） // 3. 记录TCC日志 tccLogRepository.saveTryLog(order); } // 第二阶段：Confirm（确认）- 幂等操作 @Transactional public void confirmPhase(String transactionId) { // 1. 检查是否已执行过Confirm if (tccLogRepository.isConfirmed(transactionId)) { return; } // 2. 执行真正的业务操作 // 库存服务：扣减冻结的库存 // 账户服务：扣除冻结的余额 // 订单服务：更新订单状态为CONFIRMED // 3. 更新TCC日志状态为CONFIRMED tccLogRepository.updateToConfirmed(transactionId); } // 第二阶段：Cancel（取消）- 幂等操作 @Transactional public void cancelPhase(String transactionId) { // 1. 检查是否已执行过Cancel if (tccLogRepository.isCancelled(transactionId)) { return; } // 2. 执行补偿操作（逆向操作） // 库存服务：释放冻结的库存 // 账户服务：释放冻结的余额 // 订单服务：更新订单状态为CANCELLED // 3. 更新TCC日志状态为CANCELLED tccLogRepository.updateToCancelled(transactionId); } /** * TCC优缺点： * 优点： * - 性能好（一阶段不锁定资源） * - 最终一致性 * - 适用于长事务 * * 缺点： * - 业务侵入性强（需要拆分为3个方法） * - 需要实现幂等和空回滚 * - 开发成本高 */ }

3.4 Saga模式

/** * Saga模式 - 长事务解决方案 * 两种实现方式： * 1. 协同式（Choreography）- 事件驱动 * 2. 编排式（Orchestration）- 中央协调器 */ // 编排式Saga示例 @Service public class OrderSagaOrchestrator { @Autowired private SagaTransactionManager transactionManager; public void createOrder(OrderDTO order) { // 定义Saga执行步骤 SagaDefinition saga = SagaDefinition.builder() .step("createOrder", this::createOrderStep) .step("reserveInventory", this::reserveInventoryStep) .step("processPayment", this::processPaymentStep) .withCompensation("compensateOrder", this::compensateOrderStep) .withCompensation("releaseInventory", this::releaseInventoryStep) .withCompensation("refundPayment", this::refundPaymentStep) .build(); // 执行Saga transactionManager.execute(saga, order); } private void createOrderStep(OrderDTO order) { // 创建订单（状态为PENDING） orderService.createPendingOrder(order); } private void reserveInventoryStep(OrderDTO order) { // 预留库存 inventoryService.reserve(order.getItems()); } private void processPaymentStep(OrderDTO order) { // 处理支付 paymentService.charge(order.getUserId(), order.getAmount()); } // 补偿方法 private void compensateOrderStep(OrderDTO order) { // 取消订单 orderService.cancelOrder(order.getId()); } /** * Saga特点： * 1. 每个步骤都有对应的补偿操作 * 2. 补偿操作顺序与正向操作相反 * 3. 允许中间状态（最终一致） * 4. 适用于长业务流程 */ }

3.5 本地消息表

/** * 本地消息表 - 最终一致性方案 * 核心思想：将分布式事务拆分为多个本地事务 */ @Service public class OrderMessageService { @Transactional public void createOrderWithMessage(OrderDTO order) { // 1. 执行业务操作（本地事务） orderService.createOrder(order); // 2. 记录消息到本地消息表（同一个事务） Message message = new Message(); message.setId(UUID.randomUUID().toString()); message.setTopic("ORDER_CREATED"); message.setContent(JSON.toJSONString(order)); message.setStatus(MessageStatus.PENDING); messageRepository.save(message); // 事务提交后，消息才会被发送 } // 消息发送服务（定时任务） @Scheduled(fixedDelay = 5000) public void sendPendingMessages() { List<Message> pendingMessages = messageRepository.findByStatus(MessageStatus.PENDING); for (Message message : pendingMessages) { try { // 发送消息到MQ mqProducer.send(message.getTopic(), message.getContent()); // 更新消息状态为SENT message.setStatus(MessageStatus.SENT); messageRepository.save(message); } catch (Exception e) { log.error("发送消息失败: {}", message.getId(), e); // 记录重试次数，超过阈值人工处理 message.incrementRetryCount(); messageRepository.save(message); } } } // 消息消费端（保证幂等性） @RabbitListener(queues = "ORDER_CREATED_QUEUE") public void handleOrderCreated(String messageBody) { // 1. 解析消息 OrderDTO order = JSON.parseObject(messageBody, OrderDTO.class); // 2. 检查是否已处理（防重表） if (processedMessageRepository.existsById(order.getId())) { return; // 已处理，直接返回 } // 3. 执行业务（如扣减库存） try { inventoryService.deductStock(order.getItems()); // 4. 记录已处理的消息 processedMessageRepository.save(new ProcessedMessage(order.getId())); } catch (Exception e) { // 消费失败，MQ会自动重试 throw new RuntimeException("处理消息失败", e); } } }

3.6 最大努力通知

/** * 最大努力通知 - 适用于对一致性要求不高的场景 */ @Service public class BestEffortNotificationService { private static final int MAX_RETRY = 5; private static final long[] RETRY_INTERVALS = {1000, 5000, 10000, 30000, 60000}; public void processWithNotification(OrderDTO order) { // 1. 执行业务操作 orderService.createOrder(order); // 2. 异步发送通知（不保证一定成功） CompletableFuture.runAsync(() -> { sendNotificationWithRetry(order); }); } private void sendNotificationWithRetry(OrderDTO order) { int retryCount = 0; while (retryCount < MAX_RETRY) { try { // 发送通知（如短信、邮件） notificationService.sendOrderCreatedNotification(order); log.info("通知发送成功: {}", order.getId()); return; } catch (Exception e) { retryCount++; log.warn("通知发送失败，准备重试 {}/{}", retryCount, MAX_RETRY, e); if (retryCount < MAX_RETRY) { try { Thread.sleep(RETRY_INTERVALS[retryCount - 1]); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); } } } } // 超过最大重试次数，记录日志人工处理 log.error("通知发送失败，超过最大重试次数: {}", order.getId()); alertService.sendAlert("通知发送失败，需要人工处理: " + order.getId()); } }

3.7 Seata框架使用示例

/** * Seata - 开源的分布式事务解决方案 * 支持AT、TCC、Saga、XA多种模式 */ // AT模式（自动补偿）示例 @Configuration public class SeataConfig { @Bean public DataSource dataSource(DataSourceProperties properties) { // 1. 配置数据源代理 DruidDataSource druidDataSource = new DruidDataSource(); // ... 配置数据源 // 2. 使用Seata的数据源代理 return new DataSourceProxy(druidDataSource); } } // 业务代码 @Service public class OrderService { // 使用@GlobalTransactional注解开启全局事务 @GlobalTransactional(timeoutMills = 300000, name = "createOrder") public void createOrder(OrderDTO order) { // 1. 创建订单（本地事务） orderMapper.insert(order); // 2. 扣减库存（远程服务） inventoryFeignClient.deduct(order.getItems()); // 3. 扣减余额（远程服务） accountFeignClient.decrease(order.getUserId(), order.getAmount()); // 所有操作要么全部成功，要么全部回滚 } } // Seata AT模式原理： // 1. 一阶段：执行业务SQL，保存undo_log（前后镜像） // 2. 二阶段提交：删除undo_log // 3. 二阶段回滚：根据undo_log恢复数据

四、实际业务场景详解

4.1 电商下单场景

/** * 电商下单完整分布式事务方案 * 需求：创建订单 → 扣减库存 → 扣减余额 → 生成积分 */ @Service public class EcommerceOrderService { // 方案1：TCC模式（强一致性要求） public void createOrderWithTCC(OrderDTO order) { // 1. Try阶段：资源预留 orderService.tryCreateOrder(order); // 创建待确认订单 inventoryService.tryDeductStock(order); // 冻结库存 accountService.tryDeductBalance(order); // 冻结余额 // 2. Confirm阶段：实际执行 if (allTrySuccess()) { orderService.confirmOrder(order); // 确认订单 inventoryService.confirmDeduct(order); // 扣减库存 accountService.confirmDeduct(order); // 扣减余额 pointService.grantPoints(order); // 发放积分 } else { // 3. Cancel阶段：补偿操作 orderService.cancelOrder(order); inventoryService.cancelDeduct(order); accountService.cancelDeduct(order); } } // 方案2：Saga模式（长流程、允许中间状态） public void createOrderWithSaga(OrderDTO order) { SagaExecutor saga = new SagaExecutor(); saga.addStep( "createOrder", () -> orderService.create(order), () -> orderService.cancel(order) ); saga.addStep( "deductInventory", () -> inventoryService.deduct(order.getItems()), () -> inventoryService.restore(order.getItems()) ); saga.addStep( "processPayment", () -> paymentService.charge(order), () -> paymentService.refund(order) ); // 执行Saga，任一失败则执行补偿 saga.execute(); } // 方案3：本地消息表（最终一致性，解耦） @Transactional public void createOrderWithLocalMessage(OrderDTO order) { // 1. 创建订单（本地事务） orderService.create(order); // 2. 记录本地消息（同一个事务） localMessageService.saveMessage( "ORDER_CREATED", order, Arrays.asList( new MessageTarget("inventory", "deduct"), new MessageTarget("account", "deduct"), new MessageTarget("point", "grant") ) ); // 3. 异步处理器会发送消息到各个服务 } }

4.2 金融转账场景

/** * 银行转账场景 * 需求：A账户扣款 → B账户加款 → 记录流水 * 特点：强一致性要求，金额不能出错 */ @Service public class BankTransferService { // 方案1：XA协议（强一致性） @Transactional @JtaTransaction public void transferWithXA(String fromAccount, String toAccount, BigDecimal amount) { // 使用JTA/XA数据源，支持分布式事务 // 1. 扣减A账户余额 accountRepository.decreaseBalance(fromAccount, amount); // 2. 增加B账户余额 accountRepository.increaseBalance(toAccount, amount); // 3. 记录交易流水 transactionRepository.save(createTransaction(fromAccount, toAccount, amount)); // 三者在同一个XA事务中 } // 方案2：TCC模式（性能更好） public void transferWithTCC(String fromAccount, String toAccount, BigDecimal amount) { String txId = UUID.randomUUID().toString(); try { // Try阶段 accountService.tryDecrease(fromAccount, amount, txId); accountService.tryIncrease(toAccount, amount, txId); // Confirm阶段 accountService.confirmDecrease(txId); accountService.confirmIncrease(txId); transactionService.record(txId, fromAccount, toAccount, amount); } catch (Exception e) { // Cancel阶段 accountService.cancelDecrease(txId); accountService.cancelIncrease(txId); throw e; } } // 方案3：对账补偿（最终一致性 + 对账） public void transferWithReconciliation(String fromAccount, String toAccount, BigDecimal amount) { // 1. 记录转账请求（唯一流水号） String transferId = recordTransferRequest(fromAccount, toAccount, amount); // 2. 异步执行转账 asyncTransferService.executeTransfer(transferId); // 3. 定时对账（补偿不一致） // 每日对账，发现不一致时人工或自动修复 } }

4.3 酒店预订场景

/** * 酒店+机票+租车套餐预订 * 需求：多服务协同，长业务流程，允许部分失败 */ @Service public class TravelBookingService { // Saga编排式方案 public BookingResult bookTravelPackage(TravelRequest request) { SagaExecution saga = SagaExecution.builder() .addStep("validateRequest", this::validateRequest) .addStep("bookHotel", this::bookHotel) .addStep("bookFlight", this::bookFlight) .addStep("bookCar", this::bookCar) .addStep("makePayment", this::makePayment) .addStep("sendConfirmation", this::sendConfirmation) .build(); return saga.execute(request); } private void bookHotel(TravelRequest request) { // 酒店预订（可能有多个房型选择） HotelBooking hotelBooking = hotelService.book( request.getHotelId(), request.getCheckInDate(), request.getCheckOutDate(), request.getRooms() ); request.setHotelBooking(hotelBooking); // 补偿操作：取消酒店预订 // hotelService.cancel(hotelBooking.getId()); } private void bookFlight(TravelRequest request) { // 机票预订 FlightBooking flightBooking = flightService.book( request.getFlightNumber(), request.getPassengers() ); request.setFlightBooking(flightBooking); } // 降级策略：部分服务失败时的处理 private BookingResult handlePartialFailure(TravelRequest request, Exception e) { if (request.getHotelBooking() != null && request.getFlightBooking() == null) { // 酒店已订，机票失败 → 询问用户是否继续 return BookingResult.partialSuccess( "酒店预订成功，机票预订失败。是否继续？", request.getHotelBooking() ); } // 其他失败情况... return BookingResult.failed("预订失败: " + e.getMessage()); } }

五、分布式事务选型指南

5.1 方案对比矩阵

5.2 选型决策树

public class DistributedTransactionSelector { public TransactionStrategy selectStrategy(BusinessContext context) { // 1. 强一致性要求？ if (context.isStrongConsistencyRequired()) { // 性能要求高？ if (context.isHighPerformanceRequired()) { return TransactionStrategy.TCC; // TCC模式 } else { return TransactionStrategy.XA; // XA/2PC } } // 2. 长业务流程？ if (context.isLongRunningProcess()) { // 需要明确补偿？ if (context.needExplicitCompensation()) { return TransactionStrategy.SAGA; // Saga模式 } else { return TransactionStrategy.LOCAL_MESSAGE; // 本地消息表 } } // 3. 异步解耦场景？ if (context.isAsyncDecoupling()) { return TransactionStrategy.LOCAL_MESSAGE; // 本地消息表 } // 4. 最终一致性即可？ if (context.acceptEventualConsistency()) { // 允许丢失？ if (context.tolerateMessageLoss()) { return TransactionStrategy.BEST_EFFORT; // 最大努力通知 } else { return TransactionStrategy.LOCAL_MESSAGE; // 本地消息表 } } // 默认：本地消息表 return TransactionStrategy.LOCAL_MESSAGE; } }

5.3 混合方案实践

/** * 实际项目中常采用混合方案 */ @Service public class HybridTransactionService { // 核心业务用TCC，边缘业务用消息队列 public void processOrder(OrderDTO order) { // 1. 核心操作使用TCC（强一致性） tccTransactionService.tryOrder(order); // 2. 非核心操作使用消息队列（最终一致性） if (tccTransactionService.isAllTrySuccess()) { // 发送异步消息 messageQueue.send("order.confirmed", order); // 确认TCC tccTransactionService.confirmOrder(order); } else { // 取消TCC tccTransactionService.cancelOrder(order); } } // 本地消息表 + 定时任务补偿 public void syncDataToDataWarehouse(DataSyncRequest request) { // 1. 业务操作 + 记录消息 businessService.process(request); messageService.saveSyncMessage(request); // 2. 定时任务会读取消息并同步到数据仓库 // 3. 失败重试 + 人工补偿兜底 } }

六、最佳实践与注意事项

6.1 幂等性设计

@Component public class IdempotentService { // 方法1：唯一业务ID public void processOrder(String orderId, String businessId) { // 检查是否已处理 if (idempotentChecker.isProcessed(businessId)) { return; // 幂等返回 } // 执行业务逻辑 doProcessOrder(orderId); // 记录已处理 idempotentChecker.markProcessed(businessId); } // 方法2：数据库唯一索引 @Table(uniqueConstraints = { @UniqueConstraint(columnNames = {"order_id", "type"}) }) public class TransactionRecord { // 插入时如果重复会抛出异常 } // 方法3：Token机制 public void processWithToken(String orderId) { // 1. 获取Token String token = tokenService.getToken(orderId); // 2. 执行业务（携带Token） orderService.process(orderId, token); // 3. 服务端验证Token // if (!tokenService.validate(token)) return; } }

6.2 空回滚与防悬挂

/** * TCC中的特殊问题处理 */ @Service public class TCCProblemHandler { // 空回滚：Try未执行，但收到了Cancel public void handleEmptyRollback(String txId) { // 检查Try是否执行过 if (!tryLogRepository.existsByTxId(txId)) { // 记录空回滚日志 emptyRollbackLogRepository.save(txId); return; // 直接返回，不执行补偿逻辑 } // 正常执行Cancel doCancel(txId); } // 防悬挂：Cancel比Try先执行 public void handleHanging(String txId) { // 检查是否已有空回滚记录 if (emptyRollbackLogRepository.existsByTxId(txId)) { // 已有空回滚记录，丢弃Try请求 return; } // 正常执行Try doTry(txId); } }

6.3 监控与告警

@Configuration public class TransactionMonitorConfig { @Bean public MeterRegistryCustomizer<MeterRegistry> metricsCommonTags() { return registry -> registry.config().commonTags( "application", "order-service", "transaction_type", "distributed" ); } // 监控指标 @Component public class TransactionMetrics { private final Counter successCounter; private final Counter failureCounter; private final Timer transactionTimer; public void recordTransaction(String type, long duration, boolean success) { if (success) { successCounter.increment(); } else { failureCounter.increment(); } transactionTimer.record(duration, TimeUnit.MILLISECONDS); // 发送到监控系统 monitorService.recordTransaction(type, duration, success); } } // 告警规则 @Component public class TransactionAlert { @Scheduled(fixedRate = 60000) // 每分钟检查 public void checkTransactionHealth() { // 1. 检查失败率 double failureRate = calculateFailureRate(); if (failureRate > 0.05) { // 失败率超过5% alertService.send("交易失败率异常: " + failureRate); } // 2. 检查平均耗时 long avgDuration = calculateAverageDuration(); if (avgDuration > 5000) { // 平均耗时超过5秒 alertService.send("交易耗时异常: " + avgDuration + "ms"); } // 3. 检查死锁 checkDeadTransactions(); } } }

七、常见问题与解决方案

7.1 热点数据问题

@Service public class HotspotSolution { // 问题：库存扣减热点 // 解决方案1：库存分段 public boolean deductStockWithSegment(String productId, int quantity) { // 将库存分为多个段 List<StockSegment> segments = stockSegmentService.getSegments(productId); // 随机或轮询选择段 StockSegment segment = selectSegment(segments); // 扣减分段库存 return stockSegmentService.deduct(segment.getId(), quantity); } // 解决方案2：库存预扣+异步同步 public boolean deductStockWithBuffer(String productId, int quantity) { // 1. 扣减缓冲库存（Redis） boolean success = redisStockService.deduct(productId, quantity); if (success) { // 2. 异步同步到数据库 asyncStockService.syncToDB(productId, quantity); return true; } return false; } }

7.2 数据一致性验证

@Service public class ConsistencyValidator { // 定时对账任务 @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点 public void dailyReconciliation() { // 1. 获取需要核对的数据 List<Order> orders = orderService.getYesterdayOrders(); for (Order order : orders) { // 2. 核对各个系统数据 boolean inventoryMatch = checkInventory(order); boolean accountMatch = checkAccount(order); boolean pointMatch = checkPoints(order); // 3. 记录不一致 if (!(inventoryMatch && accountMatch && pointMatch)) { reconciliationService.recordMismatch(order); } } // 4. 发送对账报告 sendReconciliationReport(); } // 实时一致性检查 public void realtimeConsistencyCheck(String orderId) { // 使用事件溯源模式 List<DomainEvent> events = eventStore.getEvents(orderId); // 重新聚合状态 OrderState state = replayEvents(events); // 与当前状态对比 Order currentState = orderService.getOrder(orderId); if (!state.equals(currentState)) { // 触发修复流程 consistencyFixer.fixInconsistency(orderId); } } }

总结

分布式事务的选择和实现需要根据具体的业务场景、一致性要求和性能需求来决定：

1. 强一致性场景（金融、交易）：优先考虑XA/TCC

2. 长业务流程（旅行、保险）：适合Saga模式

3. 最终一致性（电商、社交）：本地消息表是良好选择

4. 弱一致性（日志、通知）：最大努力通知即可

实际项目中，通常采用混合策略，核心链路用强一致性方案，非核心链路用最终一致性方案。同时，完善的监控、告警、对账和补偿机制是分布式事务系统稳定运行的保障。