当前位置：首页 > news >正文

分布式事务seata框架下at/tcc/sage模式代码

news 2026/8/15 23:14:05

下载 seata: https://github.com/apache/incubator-seata/releases#release-v2.6.0

启动seata: window执行 seata-server.bat,linux执行 seata-server.sh

AT模式代码:

创建订单模块,账户模块,库存模块,每个模块有自己的数据库,订单模块调另外两个模块,每个模块要建立undo_log表用于回滚

package com.example.order.service;import com.example.order.entity.Order;
import com.example.order.mapper.OrderMapper;
import io.seata.spring.annotation.GlobalTransactional;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;import java.util.HashMap;
import java.util.Map;/*** 下单服务。* <p>* 这里的 @GlobalTransactional 就是「TM（事务管理器）」的标志：* 它负责开启 / 结束全局事务，并向 TC 发起全局提交或全局回滚。* 同时本服务也操作自己的订单库，所以它同时也是「RM（资源管理器）」。*/
@Service
@RequiredArgsConstructor
public class OrderService {private final OrderMapper orderMapper;private final RestTemplate restTemplate;@Value("${remote.storage-url}")private String storageUrl;@Value("${remote.account-url}")private String accountUrl;/*** 下单：创建订单 + 扣库存 + 扣余额，三步必须一起成功，否则一起回滚。* 注意：这里不需要 @Transactional，@GlobalTransactional 已接管整个链路。*/@GlobalTransactional(name = "create-order-tx", rollbackFor = Exception.class)public void createOrder(Order order) {// ① 本地分支事务：创建订单（本服务 = RM，操作 seata_order 库）
        orderMapper.insert(order);// ② 远程分支事务：扣库存（storage-service = RM，操作 seata_storage 库）Map<String, Object> storageParam = new HashMap<>();storageParam.put("productId", order.getProductId());storageParam.put("count", order.getCount());restTemplate.postForObject(storageUrl + "/storage/deduct", storageParam, String.class);// ③ 远程分支事务：扣余额（account-service = RM，操作 seata_account 库）Map<String, Object> accountParam = new HashMap<>();accountParam.put("userId", order.getUserId());accountParam.put("money", order.getMoney());restTemplate.postForObject(accountUrl + "/account/deduct", accountParam, String.class);// 三步全部成功，方法正常返回 → TM 通知 TC 全局提交// 任何一步抛异常（如库存不足、余额不足）→ 异常向上传播 → TM 通知 TC 全局回滚
    }
}

TC (事务协调器)：独立的服务端，是全局的决策中心，负责协调和管理所有分支事务。就是单独启动的seata服务
TM (事务管理器)：嵌入在发起全局事务的应用中，负责开启、提交或回滚全局事务。就是加了@GlobalTransactional注解的服务
RM (资源管理器)：嵌入在每个参与事务的应用中，负责执行本地事务，并向 TC 注册分支事务和上报状态。每个分支事务

　　重要概念XID:

在 Seata 中，XID 是一个全局唯一的事务标识符,当 TM（事务管理器）向 TC（事务协调器）发起开启全局事务请求时，TC 会生成这个 XID，并返回给发起方。后续无论是 TM 提交/回滚全局事务，还是 RM 注册分支事务，都需要携带这个 XID，TC 通过它来唯一识别和管理这个全局事务实例。
在微服务调用链中，XID 必须透传给下游服务。发起方会将 XID 塞入调用链的上下文（如 RPC 的隐式参数、HTTP 的 Header）中。下游服务接收到请求后，会解析出 XID 并绑定到本地线程变量（RootContext）。这样，原本孤立的本地服务就被“拉入”了同一个全局事务组中，确保所有跨服务的调用都属于同一个分布式事务。
RM 在执行本地事务时，会生成 SQL 执行前后的镜像数据，并插入到数据库的 UNDO_LOG 表中。UNDO_LOG 表里会专门有一个字段记录当前的 XID。当全局事务需要回滚时，TC 发送回滚请求并带上 XID，RM 根据这个 XID 在本地数据库中查询对应的 UNDO_LOG 记录，从而精准恢复数据，而不会错误地回滚其他事务的数据。

项目中resttemplate拦截xid并传给下游服务代码

package com.example.order.config;import io.seata.core.context.RootContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpRequest;
import org.springframework.http.client.ClientHttpRequestExecution;
import org.springframework.http.client.ClientHttpRequestInterceptor;
import org.springframework.http.client.ClientHttpResponse;
import org.springframework.web.client.RestTemplate;import java.io.IOException;
import java.util.Collections;/*** RestTemplate 配置：关键点是拦截器把全局事务 XID 放进 HTTP 头，* 这样下游服务（storage/account）才能知道自己属于哪个全局事务。*/
@Configuration
public class RestTemplateConfig {@Beanpublic RestTemplate restTemplate() {RestTemplate restTemplate = new RestTemplate();restTemplate.setInterceptors(Collections.singletonList(new SeataXidInterceptor()));return restTemplate;}/** 出站拦截器：把当前线程的 XID 放到请求头 TX_XID */static class SeataXidInterceptor implements ClientHttpRequestInterceptor {@Overridepublic ClientHttpResponse intercept(HttpRequest request, byte[] body,ClientHttpRequestExecution execution) throws IOException {String xid = RootContext.getXID();if (xid != null) {request.getHeaders().add(RootContext.KEY_XID, xid);}return execution.execute(request, body);}}
}

order服务调下游服务获取xid时的断点

此时order库的undo_log表

seata下的at模式每个分布式事务都先执行各自的提交,如果有RM需要回滚,TC再通知所有RM统一回滚,如果不需要回滚就通知RM删除undo_log数据

TCC模式

对订单/库存/账户表的变更:订单表添加status,库存/账户表添加freeze

每个库都添加tcc_fence_log表,用于幂等/防悬挂

CREATE TABLE `t_order` (
`id` bigint NOT NULL AUTO_INCREMENT,
`user_id` bigint NOT NULL COMMENT '用户ID',
`product_id` bigint NOT NULL COMMENT '商品ID',
`count` int NOT NULL COMMENT '购买数量',
`money` decimal(10,2) NOT NULL COMMENT '订单金额',
`status` int NOT NULL DEFAULT '0' COMMENT '订单状态 0=INIT(已预留) 1=CONFIRMED 2=CANCELLED',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='订单表'

CREATE TABLE `t_storage` (
`id` bigint NOT NULL AUTO_INCREMENT,
`product_id` bigint NOT NULL COMMENT '商品ID',
`total` int NOT NULL COMMENT '总库存',
`used` int NOT NULL DEFAULT '0' COMMENT '已用库存',
`residue` int NOT NULL COMMENT '剩余库存',
`freeze` int NOT NULL DEFAULT '0' COMMENT '冻结库存（TCC Try 预留，Confirm 转 used，Cancel 退回 residue）',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='库存表';

CREATE TABLE `t_account` (
`id` bigint NOT NULL AUTO_INCREMENT,
`user_id` bigint NOT NULL COMMENT '用户ID',
`total` decimal(10,2) NOT NULL COMMENT '总额度',
`used` decimal(10,2) NOT NULL DEFAULT '0.00' COMMENT '已用额度',
`residue` decimal(10,2) NOT NULL COMMENT '剩余额度',
`freeze` decimal(10,2) NOT NULL DEFAULT '0.00' COMMENT '冻结余额（TCC Try 预留，Confirm 转 used，Cancel 退回 residue）',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='账户表'

CREATE TABLE `tcc_fence_log` (
`xid` varchar(128) NOT NULL COMMENT '全局事务ID',
`branch_id` bigint NOT NULL COMMENT '分支事务ID',
`action_name` varchar(64) NOT NULL COMMENT '分支动作名（@TwoPhaseBusinessAction.name）',
`status` tinyint NOT NULL COMMENT '状态 1=try 2=confirm 3=cancel 4=悬挂',
`gmt_create` datetime(3) NOT NULL COMMENT '创建时间',
`gmt_modified` datetime(3) NOT NULL COMMENT '修改时间',
PRIMARY KEY (`xid`,`branch_id`),
KEY `idx_gmt_modified` (`gmt_modified`),
KEY `idx_status` (`status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='Seata TCC 防悬挂/幂等日志表'

订单模块下单,远程调用库存/账户模块扣减库存和余额

  @GlobalTransactional(name = "create-order-tcc-tx", rollbackFor = Exception.class)public void createOrder(Order order) {// ① 本地 TCC 分支：预下单（status=0）orderTccAction.prepare(null, order);// ② 远程 TCC 分支：冻结库存Map<String, Object> storageParam = new HashMap<>();storageParam.put("productId", order.getProductId());storageParam.put("count", order.getCount());restTemplate.postForObject(storageUrl + "/storage/tcc/prepare", storageParam, String.class);// ③ 远程 TCC 分支：冻结余额Map<String, Object> accountParam = new HashMap<>();accountParam.put("userId", order.getUserId());accountParam.put("money", order.getMoney());restTemplate.postForObject(accountUrl + "/account/tcc/prepare", accountParam, String.class);}

@GlobalTransactional标识订单模块是TM和RM,name用于seata-service记录日志,可以省略不写
order模块代码:

/*** 订单 TCC 接口。* <p>* {@link LocalTCC} 让 Seata 自动为该接口的实现类生成代理；* {@link TwoPhaseBusinessAction} 声明 Try 是 prepare，Confirm 是 commit，Cancel 是 rollback。* name 全局唯一，是分支资源标识。* useTCCFence=true 开启 tcc_fence_log 防悬挂/幂等。*/
@LocalTCC
public interface OrderTccAction {@TwoPhaseBusinessAction(name = "orderTccAction", commitMethod = "commit",rollbackMethod = "rollback", useTCCFence = true)boolean prepare(BusinessActionContext actionContext, Order order);boolean commit(BusinessActionContext actionContext);boolean rollback(BusinessActionContext actionContext);
}

package com.example.tccorder.tcc;
@Service
@RequiredArgsConstructor
public class OrderTccActionImpl implements OrderTccAction {private final OrderMapper orderMapper;/** Try：预下单，订单状态置 0（预留），不真正确认。 */@Override@Transactional(rollbackFor = Exception.class)public boolean prepare(BusinessActionContext actionContext, Order order) {order.setStatus(0);orderMapper.insert(order);// 关键：把回填的订单 ID 塞进上下文，供 commit/rollback 更新状态用。actionContext.addActionContext("orderId", String.valueOf(order.getId()));return true;}/** Confirm：全局提交时，把订单状态置 1。 */@Override@Transactional(rollbackFor = Exception.class)public boolean commit(BusinessActionContext actionContext) {Long orderId = toLong(actionContext.getActionContext("orderId"));orderMapper.updateStatus(orderId, 1);return true;}/** Cancel：全局回滚时，把订单状态置 2（逻辑取消，保留记录）。 */@Override@Transactional(rollbackFor = Exception.class)public boolean rollback(BusinessActionContext actionContext) {Long orderId = toLong(actionContext.getActionContext("orderId"));orderMapper.updateStatus(orderId, 2);return true;}private Long toLong(Object value) {return value == null ? null : Long.valueOf(String.valueOf(value));}
}

库存模块代码:

package com.example.tccstorage.tcc;

@LocalTCC
public interface StorageTccAction {@TwoPhaseBusinessAction(name = "storageTccAction", commitMethod = "commit",rollbackMethod = "rollback", useTCCFence = true)boolean prepare(BusinessActionContext actionContext,@BusinessActionContextParameter(paramName = "productId") Long productId,@BusinessActionContextParameter(paramName = "count") Integer count);boolean commit(BusinessActionContext actionContext);boolean rollback(BusinessActionContext actionContext);
}

@LocalTCC标识的接口,会向seata注册分支事务,成为RM,

@TwoPhaseBusinessAction的name表示分支事务名,不可重复,commitMethod /rollbackMethod 标识提交和回滚时执行的方法
useTCCFence 是 @TwoPhaseBusinessAction 的一个开关属性，默认 false。设为 true 就是开启 Seata 的「TCC 防悬挂/幂等」机制，这个机制靠 tcc_fence_log 表来工作

防悬挂/幂等/空回滚逻辑不可自定义,都是seata靠状态字段完成的,如果不符合自己的业务,就设成false,自己写逻辑保证幂等tcc的好处是可以支持无事务的库,因为回滚/提交是靠自己写的commit/rollback接口完成.缺点是不能对旧项目进行改造,需要在一开始就建好冻结字段.

saga:

每个子分支必须写一个正向操作和一个对应的补偿操作,比如连续的正向操作是z1,z2,z3,对应的补偿操作是b1,b2,b3,如果z3时出错,就执行b2,b1进行补偿.

正向操作和补偿操作执行流程写在状态机配置文件中,是一个json文件,类似工作流.

创建订单代码:

package com.example.saga.controller;import com.example.saga.dto.OrderRequest;
import io.seata.saga.engine.StateMachineEngine;
import io.seata.saga.statelang.domain.ExecutionStatus;
import io.seata.saga.statelang.domain.StateMachineInstance;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;import java.util.HashMap;
import java.util.Map;
import java.util.UUID;@RestController
@RequiredArgsConstructor
public class SagaController {private final StateMachineEngine stateMachineEngine;@PostMapping("/order/create")public String create(@RequestBody OrderRequest request) {// 业务主键：既用于幂等，也存进状态机实例表String businessKey = UUID.randomUUID().toString();Map<String, Object> params = new HashMap<>();params.put("businessKey", businessKey);params.put("userId", request.getUserId());params.put("productId", request.getProductId());params.put("count", request.getCount());params.put("money", request.getMoney());// 同步执行状态机，返回最终实例StateMachineInstance inst = stateMachineEngine.startWithBusinessKey("orderSaga", null, businessKey, params);//orderSaga 对应状态机配置文件的name,表示开始的节点

if (ExecutionStatus.SU.equals(inst.getStatus()) && inst.getCompensationStatus() == null) { return "下单成功（saga），businessKey=" + businessKey; } else if (ExecutionStatus.SU.equals(inst.getCompensationStatus())) { return "下单失败，已成功补偿，businessKey=" + businessKey; } else { return "下单结果未知，businessKey=" + businessKey + "，status=" + inst.getStatus() + "，compensationStatus=" + inst.getCompensationStatus(); } } }

{"Name": "orderSaga",//代码里配置的入口"Comment": "下单 saga：创建订单 -> 扣库存 -> 扣余额，失败自动逆序补偿","StartState": "CreateOrder","Version": "0.0.1","States": {"CreateOrder": {"Type": "ServiceTask","ServiceName": "orderAction",//表示类名"ServiceMethod": "createOrder",//方法名"CompensateState": "CompensateCreateOrder",//对应的补偿"Next": "ReduceStock",//如果执行成功下一步"Input": ["$.[businessKey]","$.[userId]","$.[userId]","$.[productId]","$.[count]","$.[money]"],"Output": {"createOrderResult": "$.#root"},"Status": {"#root == true": "SU","#root == false": "FA","$Exception{java.lang.Throwable}": "UN"}},"ReduceStock": {"Type": "ServiceTask","ServiceName": "storageAction","ServiceMethod": "reduceStock","CompensateState": "CompensateReduceStock","Next": "ReduceBalance","Input": ["$.[businessKey]","$.[productId]","$.[count]"],"Output": {"reduceStockResult": "$.#root"},"Status": {"#root == true": "SU","#root == false": "FA","$Exception{java.lang.Throwable}": "UN"},"Catch": [{"Exceptions": ["java.lang.Throwable"],"Next": "CompensationTrigger"}]},"ReduceBalance": {"Type": "ServiceTask","ServiceName": "accountAction","ServiceMethod": "reduceBalance","CompensateState": "CompensateReduceBalance","Next": "Succeed","Input": ["$.[businessKey]","$.[userId]","$.[money]"],"Output": {"reduceBalanceResult": "$.#root"},"Status": {"#root == true": "SU","#root == false": "FA","$Exception{java.lang.Throwable}": "UN"},"Catch": [{"Exceptions": ["java.lang.Throwable"],"Next": "CompensationTrigger"}]},"CompensateCreateOrder": {"Type": "ServiceTask","ServiceName": "orderAction","ServiceMethod": "cancelOrder","Input": ["$.[businessKey]"]},"CompensateReduceStock": {"Type": "ServiceTask","ServiceName": "storageAction","ServiceMethod": "restoreStock","Input": ["$.[businessKey]","$.[productId]","$.[count]"]},"CompensateReduceBalance": {"Type": "ServiceTask","ServiceName": "accountAction","ServiceMethod": "restoreBalance","Input": ["$.[businessKey]","$.[userId]","$.[money]"]},"CompensationTrigger": {"Type": "CompensationTrigger","Next": "Fail"},"Succeed": {"Type": "Succeed"},"Fail": {"Type": "Fail","ErrorCode": "ORDER_FAILED","Message": "下单失败，已触发补偿"}}
}