🌐 分布式事务实战

// MySQL 原生支持 XA 协议(2PC 的标准实现) // 场景:订单库 + 库存库 跨库事务 // ===== 资源 1:订单库的 XA 连接 ===== XAConnection xaConn1 = dataSource1.getXAConnection(); XAResource xaRes1 = xaConn1.getXAResource(); Connection conn1 = xaConn1.getConnection(); // ===== 资源 2:库存库的 XA 连接 ===== XAConnection xaConn2 = dataSource2.getXAConnection(); XAResource xaRes2 = xaConn2.getXAResource(); Connection conn2 = xaConn2.getConnection(); // ⚠️ 生成全局事务 ID byte[] gtrid = "order-tx-001".getBytes(); Xid xid1 = new MyXid(gtrid, "b1".getBytes()); Xid xid2 = new MyXid(gtrid, "b2".getBytes()); try { // ===== 阶段 1:Prepare ===== xaRes1.start(xid1, XAResource.TMNOFLAGS); conn1.executeUpdate("INSERT INTO orders ..."); xaRes1.end(xid1, XAResource.TMSUCCESS); xaRes2.start(xid2, XAResource.TMNOFLAGS); conn2.executeUpdate("UPDATE stock SET qty=qty-1 ..."); xaRes2.end(xid2, XAResource.TMSUCCESS); // ⚠️ 问两个资源:能否提交? int r1 = xaRes1.prepare(xid1); int r2 = xaRes2.prepare(xid2); // ===== 阶段 2:Commit 或 Rollback ===== if (r1 == XAResource.XA_OK && r2 == XAResource.XA_OK) { xaRes1.commit(xid1, true); // 两阶段提交 true xaRes2.commit(xid2, true); } else { xaRes1.rollback(xid1); // 任何一个失败,全部回滚 xaRes2.rollback(xid2); } } catch (Exception e) { // ⚠️ 协调者挂了,参与者会一直阻塞在 prepare 状态(致命缺点) }

// ===== 1. 定义 TCC 接口(用 @LocalTCC 注解)===== @LocalTCC @TwoPhaseBusinessAction(name = "deductAccount", commitMethod = "confirm", rollbackMethod = "cancel") public interface AccountTccAction { // ⚠️ Try 阶段:冻结余额(不真扣) @BusinessActionContextParameter(paramName = "userId") @BusinessActionContextParameter(paramName = "amount") boolean prepare(BusinessActionContext ctx); // ⚠️ Confirm 阶段:真扣(把冻结的减掉) boolean confirm(BusinessActionContext ctx); // ⚠️ Cancel 阶段:解冻(把冻结的还回去) boolean cancel(BusinessActionContext ctx); } // ===== 2. 实现 ===== @Service public class AccountTccActionImpl implements AccountTccAction { @Autowired private AccountMapper accountMapper; @Autowired private FreezeMapper freezeMapper; @Override public boolean prepare(BusinessActionContext ctx) { Long userId = ctx.getActionContext("userId", Long.class); BigDecimal amount = ctx.getActionContext("amount", BigDecimal.class); // ⚠️ Try:冻结余额。balance 减,freeze 加 int rows = accountMapper.freeze(userId, amount); if (rows == 0) throw new RuntimeException("余额不足"); // Try 失败 return true; } @Override public boolean confirm(BusinessActionContext ctx) { // ⚠️ Confirm:真扣。把 freeze 里的清零 Long userId = ctx.getActionContext("userId", Long.class); BigDecimal amount = ctx.getActionContext("amount", BigDecimal.class); accountMapper.deductFreeze(userId, amount); return true; } @Override public boolean cancel(BusinessActionContext ctx) { // ⚠️ Cancel:解冻。balance 加回,freeze 清零 Long userId = ctx.getActionContext("userId", Long.class); BigDecimal amount = ctx.getActionContext("amount", BigDecimal.class); accountMapper.unfreeze(userId, amount); return true; } } // ===== 3. 业务调用 ===== @Service public class TransferService { @Autowired private AccountTccAction accountTccAction; @GlobalTransactional // ⚠️ Seata 全局事务注解 public void transfer(Long from, Long to, BigDecimal amount) { accountTccAction.prepare(null); // 扣钱方 Try accountTccAction.prepare(null); // 收钱方 Try // 两个 Try 都成功 → Seata 自动调 confirm // 任何一个失败 → Seata 自动调 cancel } }

// Saga 适合长流程:订机票 → 订酒店 → 订租车 // 任何一步失败,反向补偿前面已成功的步骤 @Component public class TravelSaga { public void bookTravel(Long userId, TravelRequest req) { // ⚠️ Saga 用状态机驱动,Seata-Saga 引擎负责编排 StateMachineEngine engine = stateMachineEngine; Map<String, Object> params = new HashMap<>(); params.put("userId", userId); params.put("flightReq", req.getFlight()); params.put("hotelReq", req.getHotel()); params.put("carReq", req.getCar()); // 启动状态机:会按定义的流程执行 // T1(bookFlight) → T2(bookHotel) → T3(bookCar) // 任何一步失败 → 反向执行补偿 // T3 失败 → C2(cancelHotel) → C1(cancelFlight) StateMachineInstance inst = engine.start( "travel-booking", // 状态机定义(JSON) BusinessType.BOOKING, params ); if (inst.getStatus() == ExecutionStatus.FA) { // 全流程失败(补偿已自动执行) throw new RuntimeException("预订失败,已自动取消"); } } } /* 状态机定义(JSON 片段): { "Name": "travel-booking", "States": [ {"Name": "BookFlight", "Type": "ServiceTask", "ServiceName": "flightService", "ServiceMethod": "book", "CompensateState": "CancelFlight"}, {"Name": "BookHotel", "Type": "ServiceTask", "ServiceName": "hotelService", "ServiceMethod": "book", "CompensateState": "CancelHotel"}, {"Name": "BookCar", "Type": "ServiceTask", "ServiceName": "carService", "ServiceMethod": "book", "CompensateState": "CancelCar"} ] } */

@Service public class OrderService { @Autowired private OrderMapper orderMapper; @Autowired private MessageTableMapper msgMapper; // 消息表 Mapper @Transactional // ⚠️ 本地事务,保证原子性 public void createOrder(Order order) { // ① 写订单表 orderMapper.insert(order); // ② 写本地消息表(同一个事务) // ⚠️ 关键:订单和消息在同一个事务,要么都成功要么都失败 Message msg = new Message(); msg.setBizId(order.getId()); msg.setType("ADD_POINT"); msg.setPayload(JSON.toJSONString(order)); msg.setStatus(0); // 0=待发送 msg.setRetryCount(0); msg.setCreateTime(new Date()); msgMapper.insert(msg); // 事务提交:订单 + 消息 一起落库 } } // ===== 定时任务:扫描消息表,发到 MQ ===== @Component public class MessageScheduler { @Scheduled(fixedDelay = 1000) // 每秒扫一次 public void sendMessages() { List<Message> pending = msgMapper.selectByStatus(0, 100); // 取 100 条 for (Message msg : pending) { try { mqProducer.send("point-topic", msg); msgMapper.updateStatus(msg.getId(), 1); // 1=已发送 } catch (Exception e) { msgMapper.incrRetry(msg.getId()); // 重试次数 +1 if (msg.getRetryCount() > 5) { msgMapper.updateStatus(msg.getId(), 2); // 2=失败,人工介入 } } } } } // ===== 积分消费者 ===== @Component @RocketMQMessageListener(topic = "point-topic", consumerGroup = "point-group") public class PointConsumer implements RocketMQListener<Message> { @Override public void onMessage(Message msg) { // ⚠️ 幂等:防止重复消费加两次积分 boolean isNew = redis.setIfAbsent( "msg:done:" + msg.getBizId(), "1", 24, HOURS ); if (!isNew) return; Order order = JSON.parseObject(msg.getPayload(), Order.class); pointService.add(order.getUserId(), order.getAmount()); } }

@Service public class OrderService { @Autowired private TransactionMQProducer producer; public void createOrder(Order order) throws Exception { Message msg = new Message( "order-topic", "create", order.getId(), JSON.toJSONString(order).getBytes() ); // ⚠️ 发送半消息 + 注册事务监听器 producer.sendMessageInTransaction(msg, null); } } // ===== 事务监听器:执行本地事务 + 回查 ===== public class OrderTxListener implements TransactionListener { @Autowired private OrderMapper orderMapper; // ① 执行本地事务(半消息发送成功后回调) @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { try { Order order = JSON.parseObject(new String(msg.getBody()), Order.class); orderMapper.insert(order); // 写订单表 return LocalTransactionState.COMMIT_MESSAGE; // 提交,消费者可见 } catch (Exception e) { return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚,消息删除 } } // ② 回查(MQ 主动问:本地事务到底成没成?) @Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { // ⚠️ 必须可靠实现,否则半消息会一直挂起 String orderId = msg.getKeys(); Order order = orderMapper.selectById(orderId); if (order != null) { return LocalTransactionState.COMMIT_MESSAGE; // 订单在,事务成 } return LocalTransactionState.ROLLBACK_MESSAGE; // 订单不在,回滚 } }

// ===== Seata-AT 模式:业务代码几乎不用改 ===== @Service public class OrderService { @Autowired private OrderMapper orderMapper; @Autowired private StockFeignClient stockFeignClient; // 调库存服务 @GlobalTransactional // ⚠️ 唯一改动:换注解 public void createOrder(Order order) { // ① 写订单(本地事务) orderMapper.insert(order); // ② 调库存服务扣库存(远程) Result r = stockFeignClient.deduct(order.getProductId(), order.getQty()); if (!r.isSuccess()) { throw new RuntimeException("扣库存失败"); // 抛异常,Seata 自动回滚 } // ③ 失败时:Seata 自动用 undo_log 表回滚订单 } } /* ===== application.yml 配置 ===== seata: enabled: true application-id: order-service tx-service-group: my_tx_group service: vgroup-mapping: my_tx_group: default grouplist: default: 127.0.0.1:8091 # Seata Server 地址 */ /* ===== 每个业务库都要建 undo_log 表(Seata 自动维护)===== CREATE TABLE undo_log ( id BIGINT AUTO_INCREMENT PRIMARY KEY, branch_id BIGINT NOT NULL, xid VARCHAR(100) NOT NULL, -- 全局事务 ID context VARCHAR(128) NOT NULL, rollback_info LONGBLOB NOT NULL, -- 回滚 SQL log_status INT NOT NULL, log_created DATETIME NOT NULL, log_modified DATETIME NOT NULL ); */ /* AT 模式原理: 1. 业务 SQL 执行前,Seata 拦截,保存执行前的数据快照(before image) 2. 执行业务 SQL 3. 保存执行后的数据快照(after image) 4. 把 before/after image 写入 undo_log 表(同一个本地事务) 5. 全局事务失败时,Seata 用 undo_log 反向生成回滚 SQL 自动回滚 代价:每条 SQL 多了 2 次查询 + 1 次写入,性能比 TCC 差 优势:业务零侵入 */