本文目录导读：
我将为您提供一个完整的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());
}
}
}
这个分布式事务实现涵盖了以下几种经典方案：
Seata AT模式
- 适合中小型项目
- 自动回滚，代码侵入小
- 适合读多写少的场景
本地消息表方案
- 最终一致性
- 实现简单，可靠性高
- 适合跨系统异步调用
TCC模式
- 需要实现Try/Confirm/Cancel三个接口
- 控制粒度细，灵活性强
- 适合资金相关的高一致性场景
XA协议
- 强一致性
- 性能开销较大
- 适合对一致性要求极高的场景
消息队列最终一致性
- 通过消息中间件解耦
- 异步处理，性能好
- 适合高并发的电商场景
选择建议：
- 要求强一致性，且业务简单：使用Seata AT模式
- 资金类业务：使用TCC模式
- 跨系统，容忍最终一致：使用消息队列方案
- 数据一致性要求极高：使用XA协议
需要我详细展开某个方案的实现吗？
