Java实现分布式事务案例

wen java案例 2026-08-04 97

本文目录导读：

Java实现分布式事务案例

项目结构
基于Seata的分布式事务实现
基于消息队列的最终一致性方案
TCC分布式事务实现
XA分布式事务实现
事务补偿框架
测试用例

我将为您提供一个完整的Java分布式事务实现案例,包括多种主流方案。

项目结构

// 项目结构
distributed-transaction-demo/
├── pom.xml
├── src/main/java/com/example/distributed/
│   ├── account/
│   │   ├── controller/
│   │   ├── service/
│   │   ├── mapper/
│   │   └── entity/
│   ├── order/
│   │   ├── controller/
│   │   ├── service/
│   │   ├── mapper/
│   │   └── entity/
│   ├── common/
│   │   ├── exception/
│   │   ├── response/
│   │   └── utils/
│   └── transaction/
│       ├── seata/
│       ├── mq/
│       └── tcc/

基于Seata的分布式事务实现

pom.xml依赖

<dependencies>
    <!-- Spring Cloud -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
    </dependency>
    <!-- Seata -->
    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
    </dependency>
    <!-- MyBatis -->
    <dependency>
        <groupId>org.mybatis.spring.boot</groupId>
        <artifactId>mybatis-spring-boot-starter</artifactId>
        <version>2.2.0</version>
    </dependency>
    <!-- MySQL -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.28</version>
    </dependency>
</dependencies>

application.yml配置

server:
  port: 8081
spring:
  application:
    name: order-service
  cloud:
    nacos:
      discovery:
        server-addr: localhost:8848
    alibaba:
      seata:
        tx-service-group: my_test_tx_group
  datasource:
    url: jdbc:mysql://localhost:3306/order_db?useUnicode=true&characterEncoding=utf8
    username: root
    password: root
    driver-class-name: com.mysql.cj.jdbc.Driver
seata:
  enabled: true
  application-id: order-service
  tx-service-group: my_test_tx_group
  config:
    type: nacos
    nacos:
      namespace: ""
      serverAddr: localhost:8848
      group: SEATA_GROUP
  registry:
    type: nacos
    nacos:
      application: seata-server
      server-addr: localhost:8848
      namespace: ""
      group: SEATA_GROUP

订单服务实现

@Service
@Slf4j
public class OrderServiceImpl implements OrderService {
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private AccountFeignClient accountFeignClient;
    @Autowired
    private ProductFeignClient productFeignClient;
    @Override
    @GlobalTransactional(name = "create-order", rollbackFor = Exception.class)
    public Order createOrder(OrderDTO orderDTO) {
        log.info("开始创建订单，订单数据：{}", orderDTO);
        // 1. 扣减库存（调用商品服务）
        boolean deductStock = productFeignClient.deductStock(
            orderDTO.getProductId(), orderDTO.getQuantity());
        if (!deductStock) {
            throw new RuntimeException("库存不足");
        }
        // 2. 扣减余额（调用账户服务）
        boolean deductBalance = accountFeignClient.deductBalance(
            orderDTO.getUserId(), orderDTO.getAmount());
        if (!deductBalance) {
            throw new RuntimeException("余额不足");
        }
        // 3. 创建本地订单
        Order order = new Order();
        order.setOrderNo(generateOrderNo());
        order.setUserId(orderDTO.getUserId());
        order.setProductId(orderDTO.getProductId());
        order.setQuantity(orderDTO.getQuantity());
        order.setAmount(orderDTO.getAmount());
        order.setStatus(1);
        order.setCreateTime(new Date());
        orderMapper.insert(order);
        log.info("订单创建成功，订单号：{}", order.getOrderNo());
        return order;
    }
    private String generateOrderNo() {
        return "ORD" + System.currentTimeMillis() + 
               String.format("%04d", (int)(Math.random() * 10000));
    }
}

账户服务实现

@Service
@Slf4j
public class AccountServiceImpl implements AccountService {
    @Autowired
    private AccountMapper accountMapper;
    @Override
    @GlobalTransactional(name = "deduct-balance", rollbackFor = Exception.class)
    public boolean deductBalance(Long userId, BigDecimal amount) {
        log.info("开始扣减余额，用户：{}，金额：{}", userId, amount);
        // 查询用户账户
        Account account = accountMapper.selectByUserId(userId);
        if (account == null) {
            throw new RuntimeException("用户不存在");
        }
        // 扣减余额
        int result = accountMapper.deductBalance(userId, amount);
        if (result == 0) {
            throw new RuntimeException("余额不足");
        }
        // 记录交易流水
        AccountTransaction transaction = new AccountTransaction();
        transaction.setUserId(userId);
        transaction.setAmount(amount.negate());
        transaction.setType(1); // 1-支出
        transaction.setCreateTime(new Date());
        accountMapper.insertTransaction(transaction);
        log.info("余额扣减成功，剩余余额：{}", 
                 account.getBalance().subtract(amount));
        return true;
    }
}

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

本地消息表方案

@Service
@Slf4j
public class MessageTransactionServiceImpl {
    @Autowired
    private MessageMapper messageMapper;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Transactional
    public void createOrderAndMessage(OrderDTO orderDTO) {
        // 1. 创建本地事务：插入订单和消息记录
        Order order = new Order();
        order.setOrderNo(generateOrderNo());
        order.setUserId(orderDTO.getUserId());
        order.setAmount(orderDTO.getAmount());
        order.setStatus(1); // 待支付
        orderMapper.insert(order);
        // 2. 创建本地消息记录
        TransactionMessage message = new TransactionMessage();
        message.setMessageId(UUID.randomUUID().toString());
        message.setOrderNo(order.getOrderNo());
        message.setContent(JSONObject.toJSONString(orderDTO));
        message.setStatus(0); // 0-待发送
        message.setRetryCount(0);
        message.setNextRetryTime(new Date());
        messageMapper.insert(message);
        log.info("订单和消息记录创建成功，订单号：{}", order.getOrderNo());
    }
    /**
     * 定时任务：扫描并发送待处理消息
     */
    @Scheduled(cron = "0/30 * * * * ?")
    @Transactional
    public void sendPendingMessages() {
        // 查询待发送的消息（状态为0-待发送 且 超时5分钟）
        List<TransactionMessage> pendingMessages = 
            messageMapper.selectPendingMessages(5, 10);
        for (TransactionMessage message : pendingMessages) {
            try {
                // 发送消息到MQ
                rabbitTemplate.convertAndSend(
                    "order.exchange", 
                    "order.route", 
                    message.getContent()
                );
                // 更新消息状态为已发送
                messageMapper.updateStatus(message.getMessageId(), 1);
                log.info("消息发送成功，消息ID：{}", message.getMessageId());
            } catch (Exception e) {
                log.error("消息发送失败，消息ID：{}", message.getMessageId(), e);
                // 更新重试次数
                messageMapper.updateRetryCount(message.getMessageId());
                // 如果超过最大重试次数，标记为失败
                if (message.getRetryCount() >= 5) {
                    messageMapper.updateStatus(message.getMessageId(), 3); // 3-失败
                }
            }
        }
    }
    /**
     * 消息补偿机制
     */
    @Scheduled(cron = "0/60 * * * * ?")
    public void compensateFailedMessages() {
        // 查询处理补偿消息
        List<TransactionMessage> compensateMessages = 
            messageMapper.selectCompensateMessages(30, 10);
        for (TransactionMessage message : compensateMessages) {
            // 查询订单状态
            Order order = orderMapper.selectByOrderNo(message.getOrderNo());
            // 如果订单已取消，则无需补偿
            if (order != null && order.getStatus() == 0) {
                continue;
            }
            // 重新发送消息
            sendPendingMessages();
        }
    }
}

RabbitMQ可靠消息最终一致性

@Configuration
@Slf4j
public class RabbitMQTransactionConfig {
    @Autowired
    private ConnectionFactory connectionFactory;
    @Bean
    public RabbitTransactionManager rabbitTransactionManager() {
        return new RabbitTransactionManager(connectionFactory);
    }
    /**
     * 消息发送服务
     */
    @Service
    public static class MessageSender {
        @Autowired
        private RabbitTemplate rabbitTemplate;
        @Autowired
        private CorrelationDataService correlationDataService;
        /**
         * 发送可靠消息
         */
        @Transactional
        public void sendReliableMessage(ReliableMessage message) {
            // 1. 保存消息到本地
            correlationDataService.saveMessage(message);
            // 2. 发送消息
            CorrelationData correlationData = new CorrelationData(message.getMessageId());
            rabbitTemplate.convertAndSend(
                "exchange", 
                "routingKey", 
                message, 
                correlationData
            );
        }
        /**
         * 发送延迟消息（用于延迟检查）
         */
        public void sendDelayMessage(String messageId, long delayTime) {
            log.info("发送延迟检查消息，消息ID：{}", messageId);
            MessageProperties properties = new MessageProperties();
            properties.setDelay(Math.toIntExact(delayTime));
            Message message = MessageBuilder.builder()
                .setBody(messageId.getBytes())
                .andProperties(properties)
                .build();
            rabbitTemplate.convertAndSend(
                "delay.exchange", 
                "delay.route", 
                message
            );
        }
    }
    /**
     * 消费者确认服务
     */
    @Service
    public static class MessageConsumer {
        @Autowired
        private CorrelationDataService correlationDataService;
        @RabbitListener(queues = "order.queue")
        public void consumeOrder(Message message, Channel channel) throws Exception {
            String messageId = new String(message.getBody());
            log.info("消费者收到消息ID：{}", messageId);
            try {
                // 1. 幂等性检查
                if (correlationDataService.isDuplicate(messageId)) {
                    log.warn("重复消息，消息ID：{}", messageId);
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
                // 2. 处理业务逻辑
                ReliableMessage reliableMessage = 
                    correlationDataService.getMessage(messageId);
                processBusiness(reliableMessage);
                // 3. 标记消息为已完成
                correlationDataService.markSuccess(messageId);
                // 4. 确认消息
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Exception e) {
                log.error("消息处理失败，消息ID：{}", messageId, e);
                // 重试或放入死信队列
                if (message.getMessageProperties().getRedelivered()) {
                    channel.basicReject(
                        message.getMessageProperties().getDeliveryTag(), false);
                } else {
                    channel.basicNack(
                        message.getMessageProperties().getDeliveryTag(), false, true);
                }
            }
        }
        private void processBusiness(ReliableMessage message) {
            // 处理具体业务
            log.info("处理业务逻辑：{}", message.getBusinessType());
        }
    }
}

TCC分布式事务实现

TCC通用接口

public interface TccService {
    /**
     * Try阶段
     */
    boolean tryMethod(TccContext context);
    /**
     * Confirm阶段
     */
    boolean confirmMethod(TccContext context);
    /**
     * Cancel阶段
     */
    boolean cancelMethod(TccContext context);
}

TCC事务管理器

@Component
@Slf4j
public class TccTransactionManager {
    @Autowired
    private TccTransactionDao transactionDao;
    private ThreadLocal<TccTransaction> currentTransaction = new ThreadLocal<>();
    /**
     * 开启事务
     */
    @Transactional
    public TccTransaction beginTransaction(String businessType) {
        TccTransaction transaction = new TccTransaction();
        transaction.setTxId(UUID.randomUUID().toString());
        transaction.setBusinessType(businessType);
        transaction.setStatus(0); // 0-进行中
        transaction.setCreateTime(new Date());
        currentTransaction.set(transaction);
        transactionDao.insert(transaction);
        log.info("开启TCC事务，事务ID：{}", transaction.getTxId());
        return transaction;
    }
    /**
     * 注册参与者
     */
    public void registerParticipant(TccService service, TccContext context) {
        TccTransaction transaction = currentTransaction.get();
        if (transaction == null) {
            throw new RuntimeException("无活动事务");
        }
        TccParticipant participant = new TccParticipant();
        participant.setTxId(transaction.getTxId());
        participant.setServiceName(service.getClass().getName());
        participant.setTryMethod(service);
        participant.setContext(context);
        participant.setStatus(0);
        // 保存参与者信息
        transactionDao.insertParticipant(participant);
    }
    /**
     * 提交事务
     */
    @Transactional
    public void commit() {
        TccTransaction transaction = currentTransaction.get();
        if (transaction == null) {
            throw new RuntimeException("无活动事务");
        }
        // 1. 执行所有参与者的Confirm阶段
        List<TccParticipant> participants = 
            transactionDao.selectParticipants(transaction.getTxId());
        for (TccParticipant participant : participants) {
            try {
                boolean success = participant.getTryMethod()
                    .confirmMethod(participant.getContext());
                if (success) {
                    participant.setStatus(2);
                    transactionDao.updateParticipant(participant);
                } else {
                    throw new RuntimeException("Confirm阶段失败");
                }
            } catch (Exception e) {
                log.error("Confirm阶段执行失败", e);
                // 抛出异常，触发Cancel
                throw e;
            }
        }
        // 2. 更新事务状态为已提交
        transaction.setStatus(1);
        transactionDao.update(transaction);
        // 3. 清理ThreadLocal
        currentTransaction.remove();
        log.info("TCC事务提交成功，事务ID：{}", transaction.getTxId());
    }
    /**
     * 回滚事务
     */
    @Transactional
    public void rollback() {
        TccTransaction transaction = currentTransaction.get();
        if (transaction == null) {
            return;
        }
        // 1. 执行所有参与者的Cancel阶段
        List<TccParticipant> participants = 
            transactionDao.selectParticipants(transaction.getTxId());
        for (TccParticipant participant : participants) {
            try {
                participant.getTryMethod()
                    .cancelMethod(participant.getContext());
                participant.setStatus(3);
                transactionDao.updateParticipant(participant);
            } catch (Exception e) {
                log.error("Cancel阶段执行失败", e);
                // 记录失败，进行重试
            }
        }
        // 2. 更新事务状态为已回滚
        transaction.setStatus(4);
        transactionDao.update(transaction);
        // 3. 清理ThreadLocal
        currentTransaction.remove();
        log.info("TCC事务已回滚，事务ID：{}", transaction.getTxId());
    }
}

具体业务TCC实现

@Service
@Slf4j
public class AccountTccService implements TccService {
    @Autowired
    private AccountMapper accountMapper;
    @Autowired
    private FrozenAccountMapper frozenAccountMapper;
    @Override
    @Transactional
    public boolean tryMethod(TccContext context) {
        log.info("执行Try阶段，上下文：{}", context);
        // 解析业务参数
        TccAccountContext accountContext = JSONObject.parseObject(
            context.getBusinessContext(), TccAccountContext.class);
        // 1. 检查余额是否充足
        Account account = accountMapper.selectForUpdate(
            accountContext.getUserId());
        if (account.getBalance().compareTo(accountContext.getAmount()) < 0) {
            return false;
        }
        // 2. 冻结余额
        int result = accountMapper.freezeBalance(
            accountContext.getUserId(), 
            accountContext.getAmount()
        );
        if (result == 0) {
            return false;
        }
        // 3. 记录冻结记录
        FrozenAccount frozenAccount = new FrozenAccount();
        frozenAccount.setUserId(accountContext.getUserId());
        frozenAccount.setAmount(accountContext.getAmount());
        frozenAccount.setTxId(context.getTxId());
        frozenAccount.setStatus(1); // 1-冻结中
        frozenAccount.setCreateTime(new Date());
        frozenAccountMapper.insert(frozenAccount);
        log.info("余额冻结成功，用户：{}，金额：{}", 
                 accountContext.getUserId(), accountContext.getAmount());
        return true;
    }
    @Override
    @Transactional
    public boolean confirmMethod(TccContext context) {
        log.info("执行Confirm阶段，事务ID：{}", context.getTxId());
        // 解析业务参数
        TccAccountContext accountContext = JSONObject.parseObject(
            context.getBusinessContext(), TccAccountContext.class);
        // 1. 转账：从冻结余额中扣减
        int result = accountMapper.completeFrozenTransaction(
            accountContext.getUserId(),
            accountContext.getAmount(),
            accountContext.getTargetUserId()
        );
        if (result > 0) {
            // 2. 更新冻结记录状态
            frozenAccountMapper.updateStatus(
                context.getTxId(), 2); // 2-已完成
            log.info("转账确认成功，事务ID：{}", context.getTxId());
            return true;
        }
        return false;
    }
    @Override
    @Transactional
    public boolean cancelMethod(TccContext context) {
        log.info("执行Cancel阶段，事务ID：{}", context.getTxId());
        // 解析业务参数
        TccAccountContext accountContext = JSONObject.parseObject(
            context.getBusinessContext(), TccAccountContext.class);
        // 1. 解冻余额
        int result = accountMapper.unfreezeBalance(
            accountContext.getUserId(),
            accountContext.getAmount()
        );
        if (result > 0) {
            // 2. 更新冻结记录状态
            frozenAccountMapper.updateStatus(
                context.getTxId(), 3); // 3-已取消
            log.info("转账取消成功，事务ID：{}", context.getTxId());
            return true;
        }
        return false;
    }
}

XA分布式事务实现

@Configuration
@Slf4j
public class XaTransactionConfig {
    /**
     * 配置XA数据源
     */
    @Bean("orderDataSource")
    public DataSource orderDataSource() {
        MysqlXADataSource xaDataSource = new MysqlXADataSource();
        xaDataSource.setUrl("jdbc:mysql://localhost:3306/order_db");
        xaDataSource.setUser("root");
        xaDataSource.setPassword("root");
        return new AtomikosDataSourceBean() {
            {
                setUniqueResourceName("orderDataSource");
                setXaDataSource(xaDataSource);
                setMinPoolSize(5);
                setMaxPoolSize(20);
                setMaxLifetime(30000);
                setBorrowConnectionTimeout(10);
            }
        };
    }
    @Bean("accountDataSource")
    public DataSource accountDataSource() {
        MysqlXADataSource xaDataSource = new MysqlXADataSource();
        xaDataSource.setUrl("jdbc:mysql://localhost:3306/account_db");
        xaDataSource.setUser("root");
        xaDataSource.setPassword("root");
        return new AtomikosDataSourceBean() {
            {
                setUniqueResourceName("accountDataSource");
                setXaDataSource(xaDataSource);
                setMinPoolSize(5);
                setMaxPoolSize(20);
                setMaxLifetime(30000);
                setBorrowConnectionTimeout(10);
            }
        };
    }
    /**
     * 配置JTA事务管理器
     */
    @Bean(name = "atomikosTransactionManager")
    public JtaTransactionManager jtaTransactionManager() {
        return new JtaTransactionManager(
            new UserTransactionAdapter(),
            new TransactionManagerImp()
        );
    }
    /**
     * 使用全局事务
     */
    @Service
    public static class XaOrderService {
        @Autowired
        @Qualifier("orderDataSource")
        private DataSource orderDataSource;
        @Autowired
        @Qualifier("accountDataSource")
        private DataSource accountDataSource;
        @Autowired
        private PlatformTransactionManager transactionManager;
        /**
         * 跨库事务操作
         */
        public void createOrderAcrossDatabases(OrderDTO orderDTO) {
            TransactionTemplate transactionTemplate = 
                new TransactionTemplate(transactionManager);
            transactionTemplate.execute(status -> {
                try {
                    // 1. 订单库操作
                    JdbcTemplate orderJdbc = new JdbcTemplate(orderDataSource);
                    String orderSql = "INSERT INTO orders (order_no, user_id, amount) VALUES (?, ?, ?)";
                    orderJdbc.update(orderSql, 
                        generateOrderNo(), 
                        orderDTO.getUserId(), 
                        orderDTO.getAmount()
                    );
                    // 2. 账户库操作
                    JdbcTemplate accountJdbc = new JdbcTemplate(accountDataSource);
                    String accountSql = "UPDATE accounts SET balance = balance - ? WHERE user_id = ?";
                    int rows = accountJdbc.update(accountSql, 
                        orderDTO.getAmount(), 
                        orderDTO.getUserId()
                    );
                    if (rows == 0) {
                        throw new RuntimeException("账户扣款失败");
                    }
                    return true;
                } catch (Exception e) {
                    // 抛出异常会自动回滚所有事务
                    throw e;
                }
            });
        }
    }
}

事务补偿框架

@Component
@Slf4j
public class TransactionCompensator {
    @Autowired
    private CompensateRecordDao compensateRecordDao;
    @Autowired
    private RetryTemplate retryTemplate;
    /**
     * 执行带补偿的事务操作
     */
    public <T> T executeWithCompensation(
        String businessType,
        TransactionCallback<T> mainAction,
        TransactionCallback<T> compensateAction) {
        // 创建补偿记录
        String transactionId = UUID.randomUUID().toString();
        CompensateRecord record = new CompensateRecord();
        record.setTransactionId(transactionId);
        record.setBusinessType(businessType);
        record.setStatus(0); // 0-处理中
        record.setCreateTime(new Date());
        compensateRecordDao.insert(record);
        try {
            // 执行主事务
            T result = mainAction.doInTransaction();
            // 更新记录为成功
            compensateRecordDao.updateStatus(transactionId, 1);
            log.info("事务执行成功，事务ID：{}", transactionId);
            return result;
        } catch (Exception e) {
            log.error("事务执行失败，开始补偿，事务ID：{}", transactionId, e);
            // 执行补偿事务
            try {
                retryTemplate.execute(context -> {
                    compensateAction.doInTransaction();
                    return null;
                });
                // 更新记录为已补偿
                compensateRecordDao.updateStatus(transactionId, 2);
                log.info("补偿成功，事务ID：{}", transactionId);
            } catch (Exception retryException) {
                // 补偿失败，记录并加入死信队列
                compensateRecordDao.updateStatus(transactionId, 3);
                addToDeadLetterQueue(record);
                log.error("补偿失败，加入死信队列，事务ID：{}", transactionId, retryException);
            }
            throw e;
        }
    }
    /**
     * 死信队列处理（后台任务）
     */
    @Scheduled(cron = "0 */5 * * * ?")
    public void processDeadLetterQueue() {
        List<CompensateRecord> deadLetters = 
            compensateRecordDao.selectStatus(3, 10);
        for (CompensateRecord record : deadLetters) {
            log.info("处理死信队列中的补偿记录，事务ID：{}", record.getTransactionId());
            // 根据类型进行人工处理或自动处理
            handleDeadLetter(record);
        }
    }
    @FunctionalInterface
    public interface TransactionCallback<T> {
        T doInTransaction();
    }
}

测试用例

@SpringBootTest
@RunWith(SpringRunner.class)
public class DistributedTransactionTest {
    @Autowired
    private OrderServiceImpl orderService;
    @Autowired
    private MessageTransactionServiceImpl messageTransactionService;
    @Autowired
    private TccTransactionManager tccTransactionManager;
    @Autowired
    private AccountTccService accountTccService;
    /**
     * 测试Seata分布式事务
     */
    @Test
    public void testSeataTransaction() {
        // 创建订单
        OrderDTO orderDTO = new OrderDTO();
        orderDTO.setUserId(1L);
        orderDTO.setProductId(100L);
        orderDTO.setQuantity(2);
        orderDTO.setAmount(new BigDecimal("100.00"));
        try {
            Order order = orderService.createOrder(orderDTO);
            Assert.assertNotNull(order);
            Assert.assertEquals("ORD", order.getOrderNo().substring(0, 3));
            log.info("Seata分布式事务测试成功，订单号：{}", order.getOrderNo());
        } catch (Exception e) {
            log.error("Seata分布式事务测试失败", e);
            Assert.fail(e.getMessage());
        }
    }
    /**
     * 测试本地消息表方案
     */
    @Test
    public void testLocalMessageTable() {
        OrderDTO orderDTO = new OrderDTO();
        orderDTO.setUserId(2L);
        orderDTO.setProductId(200L);
        orderDTO.setQuantity(1);
        orderDTO.setAmount(new BigDecimal("50.00"));
        try {
            messageTransactionService.createOrderAndMessage(orderDTO);
            log.info("本地消息表方案测试成功");
            // 等待定时任务执行
            Thread.sleep(10000);
            // 验证消息发送状态
            List<TransactionMessage> messages = messageTransactionService
                .queryMessagesByOrderNo(orderDTO.getOrderNo());
            Assert.assertNotNull(messages);
            Assert.assertEquals(1, messages.size());
        } catch (Exception e) {
            log.error("本地消息表方案测试失败", e);
            Assert.fail(e.getMessage());
        }
    }
    /**
     * 测试TCC事务
     */
    @Test
    public void testTccTransaction() {
        // 模拟转账：从用户1转到用户2
        TccAccountContext context = new TccAccountContext();
        context.setUserId(1L);
        context.setTargetUserId(2L);
        context.setAmount(new BigDecimal("100.00"));
        TccContext tccContext = new TccContext();
        tccContext.setTxId(UUID.randomUUID().toString());
        tccContext.setBusinessContext(JSONObject.toJSONString(context));
        try {
            // 1. 开启事务
            tccTransactionManager.beginTransaction("transfer");
            // 2. 注册参与者（转账服务）
            tccTransactionManager.registerParticipant(accountTccService, tccContext);
            // 3. 执行业务逻辑
            boolean tryResult = accountTccService.tryMethod(tccContext);
            if (!tryResult) {
                // 提交失败，回滚
                tccTransactionManager.rollback();
                Assert.fail("TCC Try阶段失败");
            }
            // 4. 提交事务
            tccTransactionManager.commit();
            log.info("TCC事务测试成功");
        } catch (Exception e) {
            log.error("TCC事务测试失败", e);
            // 回滚
            tccTransactionManager.rollback();
            Assert.fail(e.getMessage());
        }
    }
    /**
     * 测试XA事务
     */
    @Test
    public void testXaTransaction() {
        XaOrderService xaOrderService = new XaOrderService();
        try {
            OrderDTO orderDTO = new OrderDTO();
            orderDTO.setUserId(3L);
            orderDTO.setAmount(new BigDecimal("200.00"));
            xaOrderService.createOrderAcrossDatabases(orderDTO);
            log.info("XA事务测试成功");
        } catch (Exception e) {
            log.error("XA事务测试失败", e);
            Assert.fail(e.getMessage());
        }
    }
}

这个分布式事务实现涵盖了以下几种经典方案：