【分布式技术专题】「架构实践于案例分析」总结和盘点目前常用分布式事务特别及问题分析（Seata-终） -阿里云开发者社区

分布式事务中间件对?与选择

tx-lcn
EasyTransaction
ByteTCC
Seata

Seata实现分布式事务

我们主要以Seata的分布式事务框架进行介绍分析，相关的并且针对于其三种模式进行分别说明介绍。

搭建Seata Server

前往github.com/seata/seata… 下载Seata安装包，本书使?Seata 1.0.0。将?录切换?Seata根?录，根据操作系统，执?对应命令，即可启动Seata Server。

Linux/Unix/Mac

sh ./bin/seata-server.sh
复制代码

Windows

bin\seata-server.bat
复制代码

启动时，也可指定参数

$ sh ./bin/seata-server.sh -p 8091 -h 127.0.0.1 -m file
复制代码

?持的参数如下表所示

Seata AT模式

?阶段：业务数据和回滚?志记录在同?个本地事务中提交，释放本地锁和连接资源。
?阶段：提交异步化，?常快速地完成。回滚通过?阶段的回滚?志进?反向补偿。

官??档

seata.io/zh-cn/docs/…

代码演示

Maven依赖

<dependency>
   <groupId>com.alibaba.cloud</groupId>
   <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
</dependency>
复制代码

配置

seata:
   tx-service-group: content-center-seata-service-group
   service:
    vgroup-mapping:
        content-center-seata-service-group: seata-cluster
        grouplist:
           seata-cluster: 127.0.0.1:8091
           disable-global-transaction: false
复制代码

配置说明

tx-service-group：事务分组，默认是 ${spring.application.name}-seata-service-group ，唯?即可。

vgroup-mapping：事务分组映射，表示 tx-service-group 对应到哪个Seata Server集群。
key是tx-service-group的值，value是集群名称，唯?即可

grouplist：集群中所包含的Seata Server的地址列表，key是vgroup-mapping中value的值，
value是Seata Server的地址列表

disable-global-transaction：是否开启全局事务开关，默认false。

在Seata1.0.0中，该配置?法正常读取，这是?个Bug，详? github.com/seata/seata… ，好在，该配置的默认值就是false，所以不影响使?。

创建Seata事务记录表

-- auto-generated definition
create table undo_log (
id bigint auto_increment comment 'increment id' primary key,
branch_id bigint not null comment 'branch transaction id',
xid varchar(100) not null comment 'global transaction id',
context varchar(128) not null comment 'undo_log context,such as serialization',
rollback_info longblob not null comment 'rollback info',
log_status int not null comment '0:normal status,1:defense status',
log_created datetime not null comment 'create datetime',
log_modified datetime not null comment 'modify datetime',
constraint ux_undo_log
unique (xid, branch_id)
) comment 'AT transaction mode undo table' charset = utf8;
复制代码

Controller代码：

private final ShareSeataService shareSeataService;
@PutMapping("/audit/seata1/{id}")
public Share auditByIdSeata1(@PathVariable Integer id, @RequestBody ShareAuditDTO auditDTO) {
     return this.shareSeataService.auditById(id, auditDTO);
}
复制代码

Service代码：

审核中心服务代码（含调用用户中心代码接口）

@Slf4j
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class ShareSeataService {
    private final ShareMapper shareMapper;
    private final UserCenterFeignClient userCenterFeignClient;
   @GlobalTransactional(rollbackFor = Exception.class)
    public Share auditById(Integer id, ShareAuditDTO auditDTO) {
        if (AuditStatusEnum.PASS.equals(auditDTO.getAuditStatusEnum())) { 
            userCenterFeignClient.addBonus(id, 50);
           // 故意抛异常，如果?户中?侧也能回滚，说明实现了分布式事务
          // throw new IllegalArgumentException("发?异常...");
        }
        this.auditByIdInDB(id, auditDTO);
        return this.shareMapper.selectByPrimaryKey(id);
   }
复制代码

审核中心服务代码（执行操作更新数据库机制）

public void auditByIdInDB(Integer id, ShareAuditDTO auditDTO) {
        Share share = Share.builder().id(id).auditStatus(auditDTO.getAuditStatusEnum().toString()).reason(auditDTO.getReason())
       .build();
        this.shareMapper.updateByPrimaryKeySelective(share);
    }
复制代码

@GlobalTransactional 注解?来创建分布式事务。

被调??代码

Maven依赖

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
</dependency>
复制代码

配置

seata:
    tx-service-group: user-center-seata-service-group
      service:
      vgroup-mapping:
         user-center-seata-service-group: seata-cluster
         grouplist:
            seata-cluster: 127.0.0.1:8091
            disable-global-transaction: false
复制代码

可以看出来，差距主要体现在tx-service-group的值。

Controller代码：

@GetMapping("/add-bonus/{id}/{bonus}")
public User addBonus(@PathVariable Integer id, @PathVariable Integer bonus) {
    this.userService.addBonus(id, bonus);
    return this.userService.findById(id);
}
复制代码

Service代码：

@Slf4j
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class UserService {
    private final UserMapper userMapper;
    private final BonusEventLogMapper bonusEventLogMapper;
    public User findById(Integer id) {
    // select * from user where id = #{id}
        return this.userMapper.selectByPrimaryKey(id);
    }
public void addBonus(Integer userId, Integer bonus) {
    // 1. 为?户加积分
    User user = this.userMapper.selectByPrimaryKey(userId);
    user.setBonus(user.getBonus() + bonus);
    this.userMapper.updateByPrimaryKeySelective(user);
    // 2. 记录?志到bonus_event_log表??
    this.bonusEventLogMapper.insert(BonusEventLog.builder().userId(userId).value(bonus).event("CONTRIBUTE").createTime(new Date()).description("投稿加积分..").build());
    log.info("积分添加完毕...");
 }
}
复制代码

Seata TCC模式

?阶段 prepare ?为
?阶段 commit 或 rollback ?为

需要实现的3个?法：

?阶段：

?于业务预处理的?法，即 Try 阶段、的?法，?如冻结?户的部分余额等等；

?阶段：

?于提交业务的?法，即 Commit ?法，?如扣除?户之前冻结的部分余额；
?于回滚业务的?法，即 Rollback ?法，?如返还之前冻结的?户余额；

官??档

seata.io/zh-cn/docs/…

代码演示

行为操作接口

@LocalTCC
public interface TccActionOne {
   @TwoPhaseBusinessAction(name = "TccActionOne", commitMethod = "commit", rollbackMethod = "rollback")
    boolean prepare(BusinessActionContext actionContext, int a);
    boolean commit(BusinessActionContext actionContext);
    boolean rollback(BusinessActionContext actionContext);
}
复制代码

接口实现类

实现类1

@Component
public class TccActionOneImpl implements TccActionOne {
    @Override
    public boolean prepare(BusinessActionContext actionContext, int a) {
        // 这?是本地玩的，也可以调?其他微服务的接?
        String xid = actionContext.getXid();
        System.out.println("TccActionOne prepare, xid:" + xid);
        return true;
    }
    @Override
    public boolean commit(BusinessActionContext actionContext) {
        // 这?是本地玩的，也可以调?其他微服务的接?
        String xid = actionContext.getXid();
        System.out.println("TccActionOne commit, xid:" + xid);
        ResultHolder.setActionOneResult(xid, "T");
        return true;
    }
    @Override
    public boolean rollback(BusinessActionContext actionContext) {
        // 这?是本地玩的，也可以调?其他微服务的接?
        String xid = actionContext.getXid();
        System.out.println("TccActionOne rollback, xid:" + xid);
        ResultHolder.setActionOneResult(xid, "R");
        return true;
    }
}
复制代码

@LocalTCC
public interface TccActionTwo {
    @TwoPhaseBusinessAction(name = "TccActionTwo", commitMethod = "commit", rollbackMethod = "rollback")
    boolean prepare(BusinessActionContext actionContext, int a);
    boolean commit(BusinessActionContext actionContext);
    boolean rollback(BusinessActionContext actionContext);
}
复制代码

实现类2

@Component
public class TccActionTwoImpl implements TccActionTwo {
       @Override
       public boolean prepare(BusinessActionContext actionContext, String b) {
              // 这?是本地玩的，也可以调?其他微服务的接?
              String xid = actionContext.getXid();
              System.out.println("TccActionTwo prepare, xid:" + xid);
              return true;
       }
       @Override
       public boolean commit(BusinessActionContext actionContext) {
              // 这?是本地玩的，也可以调?其他微服务的接?
              String xid = actionContext.getXid();
              System.out.println("TccActionTwo commit, xid:" + xid);
              ResultHolder.setActionTwoResult(xid, "T");
              return true;
       }
       @Override
       public boolean rollback(BusinessActionContext actionContext) {
              // 这?是本地玩的，也可以调?其他微服务的接?
              String xid = actionContext.getXid();
              System.out.println("TccActionTwo rollback, xid:" + xid);
              ResultHolder.setActionTwoResult(xid, "R");
              return true;
       }
}
复制代码

聚合实现服务业务实现类执行

@Service
public class ShareSeataService{
    @Autowired
    TccActionOne tccActionOne;
    @Autowired
    TccActionTwo tccActionTwo;
    @GlobalTransactional
    public void tccTransactionCommit(Map<String, String> paramMap) {
        //第一个TCC 事务参与者
        boolean result = tccActionOne.prepare(null, "one");
        if (!result) {
      paramMap.put("xid",RootContext.getXID());
            throw new RuntimeException("TccActionOne failed.");
        }
        List list = new ArrayList();
        list.add("c1");
        list.add("c2");
        result = tccActionTwo.prepare(null, "two");
        if (!result) {
            paramMap.put("xid",RootContext.getXID());
            throw new RuntimeException("TccActionTwo failed.");
        }
       paramMap.put("xid",RootContext.getXID());
        return ;
    }
}
// 回滚的代码相似，就不写了
复制代码

执行调用点

@Slf4j
@RestController
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class ShareAdminController {
    private final ShareSeataService shareSeataService;
    @GetMapping("tcc-commit")
    public String tccTransactionCommit() {
        Map<String, String> map = new HashMap<>();
        this.shareSeataService.tccTransactionCommit(map);
        String xid = map.get("xid");
        // 结果T
        return ResultHolder.getActionOneResult(xid);
    }
    @GetMapping("/tcc-rollback")
    public String tccTransactionRollback() {
       Map<String, String> map = new HashMap<>();
       try {
          this.shareSeataService.tccTransactionRollback(map);
       } catch (Throwable t) {
          log.warn("事务回滚..", t);
       }
       String xid = map.get("xid");
       // 结果R
       return ResultHolder.getActionOneResult(xid);
   }
}
复制代码

定义状态机?件：

{
    "Name": "reduceInventoryAndBalance",
    "Comment": "reduce inventory then reduce balance in a transaction",
    "StartState": "ReduceInventory",
    "Version": "0.0.1",
    "States": {
        "ReduceInventory": {
            "Type": "ServiceTask",
            "ServiceName": "inventoryAction",
            "ServiceMethod": "reduce",
            "CompensateState": "CompensateReduceInventory",
            "Next": "ChoiceState",
            "Input": [
                "$.[businessKey]",
                "$.[count]"
            ],
            "Output": {
                "reduceInventoryResult": "$.#root"
            },
            "Status": {
                "#root == true": "SU",
                "#root == false": "FA",
                "$Exception{java.lang.Throwable}": "UN"
            }
        },
        "ChoiceState": {
            "Type": "Choice",
            "Choices": [
                {
                    "Expression": "[reduceInventoryResult] == true",
                    "Next": "ReduceBalance"
                }
            ],
            "Default": "Fail"
        },
        "ReduceBalance": {
            "Type": "ServiceTask",
            "ServiceName": "balanceAction",
            "ServiceMethod": "reduce",
            "CompensateState": "CompensateReduceBalance",
            "Input": [
                "$.[businessKey]",
                "$.[amount]",
                {
                    "throwException": "$.[mockReduceBalanceFail]"分布式事务27
                }
            ],
            "Output": {
                "compensateReduceBalanceResult": "$.#root"
            },
            "Status": {
                "#root == true": "SU",
                "#root == false": "FA",
                "$Exception{java.lang.Throwable}": "UN"
            },
            "Catch": [
                {
                    "Exceptions": [
                        "java.lang.Throwable"
                    ],
                    "Next": "CompensationTrigger"
                }
            ],
            "Next": "Succeed"
        },
        "CompensateReduceInventory": {
            "Type": "ServiceTask",
            "ServiceName": "inventoryAction",
            "ServiceMethod": "compensateReduce",
            "Input": [
                "$.[businessKey]"
            ]
        },
        "CompensateReduceBalance": {
            "Type": "ServiceTask",
            "ServiceName": "balanceAction",
            "ServiceMethod": "compensateReduce",
            "Input": [
                "$.[businessKey]"
            ]
        },
        "CompensationTrigger": {
            "Type": "CompensationTrigger",
            "Next": "Fail"
        },
        "Succeed": {
            "Type": "Succeed"
        },
        "Fail": {
            "Type": "Fail",
            "ErrorCode": "PURCHASE_FAILED",
            "Message": "purchase failed"
        }
    }
}
复制代码

测试代码：

public class LocalSagaTransactionStarter {
        public static void main(String[] args) {
            AbstractApplicationContext applicationContext = new ClassPathXmlApplication
            Context(new String[] {"spring/seata-saga.xml"});
            StateMachineEngine stateMachineEngine = (StateMachineEngine) applicationCon
            text.getBean("stateMachineEngine");
            transactionCommittedDemo(stateMachineEngine);
            transactionCompensatedDemo(stateMachineEngine);
            new ApplicationKeeper(applicationContext).keep();
        }
        private static void transactionCommittedDemo(StateMachineEngine stateMachineEng
ine) {
            Map<String, Object> startParams = new HashMap<>(3);
            String businessKey = String.valueOf(System.currentTimeMillis());
            startParams.put("businessKey", businessKey);
            startParams.put("count", 10);
            startParams.put("amount", new BigDecimal("100"));
//sync test
            StateMachineInstance inst = stateMachineEngine.startWithBusinessKey("reduce
                    InventoryAndBalance", null, businessKey, startParams);
                    Assert.isTrue(ExecutionStatus.SU.equals(inst.getStatus()), "saga transactio
                            n execute failed. XID: " + inst.getId());
            System.out.println("saga transaction commit succeed. XID: " + inst.getId())
            ;
            inst = stateMachineEngine.getStateMachineConfig().getStateLogStore().getSta
            teMachineInstanceByBusinessKey(businessKey, null);
            Assert.isTrue(ExecutionStatus.SU.equals(inst.getStatus()), "saga transactio
                    n execute failed. XID: " + inst.getId());
//async test
            businessKey = String.valueOf(System.currentTimeMillis());
            inst = stateMachineEngine.startWithBusinessKeyAsync("reduceInventoryAndBala
                    nce", null, businessKey, startParams, CALL_BACK);
                    waittingForFinish(inst);
            Assert.isTrue(ExecutionStatus.SU.equals(inst.getStatus()), "saga transactio
                    n execute failed. XID: " + inst.getId());
            分布式事务
            29
            System.out.println("saga transaction commit succeed. XID: " + inst.getId())
            ;
        }
        private static void transactionCompensatedDemo(StateMachineEngine stateMachineE
ngine) {
            Map<String, Object> startParams = new HashMap<>(4);
            String businessKey = String.valueOf(System.currentTimeMillis());
            startParams.put("businessKey", businessKey);
            startParams.put("count", 10);
            startParams.put("amount", new BigDecimal("100"));
            startParams.put("mockReduceBalanceFail", "true");
//sync test
            StateMachineInstance inst = stateMachineEngine.startWithBusinessKey("reduce
                    InventoryAndBalance", null, businessKey, startParams);
//async test
                    businessKey = String.valueOf(System.currentTimeMillis());
            inst = stateMachineEngine.startWithBusinessKeyAsync("reduceInventoryAndBala
                    nce", null, businessKey, startParams, CALL_BACK);
                    waittingForFinish(inst);
            Assert.isTrue(ExecutionStatus.SU.equals(inst.getCompensationStatus()), "sag
                    a transaction compensate failed. XID: " + inst.getId());
            System.out.println("saga transaction compensate succeed. XID: " + inst.getI
                    d());
        }
        private static volatile Object lock = new Object();
        private static AsyncCallback CALL_BACK = new AsyncCallback() {
            @Override
            public void onFinished(ProcessContext context, StateMachineInstance stateMach
ineInstance) {
                synchronized (lock){
                    lock.notifyAll();
                }
            }
            @Override
            public void onError(ProcessContext context, StateMachineInstance stateMachine
Instance, Exception exp) {
                synchronized (lock){
                    lock.notifyAll();
                }
            }
        };
        private static void waittingForFinish(StateMachineInstance inst){
            synchronized (lock){
                if(ExecutionStatus.RU.equals(inst.getStatus())){
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}
复制代码

Seata saga模式

Saga 事务：最终一致性

方案简介

Saga 事务核心思想是将长事务拆分为多个本地短事务，由 Saga 事务协调器协调，如果正常结束那就正常完成，如果某个步骤失败，则根据相反顺序一次调用补偿操作。

处理流程

Saga事务基本协议

每个 Saga 事务由一系列幂等的有序子事务(sub-transaction) Ti 组成。
每个 Ti 都有对应的幂等补偿动作 Ci，补偿动作用于撤销 Ti 造成的结果。
可以看到，和 TCC 相比，Saga 没有“预留”动作，它的 Ti 就是直接提交到库。

代码实现

// 接?略
public class BalanceActionImpl implements BalanceAction {
    private static final Logger LOGGER = LoggerFactory.getLogger(BalanceActionImpl.class);
    @Override
    public boolean reduce(String businessKey, BigDecimal amount, Map<String, Object> params) {
       if(params != null && "true".equals(params.get("throwException"))){
         throw new RuntimeException("reduce balance failed");
       }
       LOGGER.info("reduce balance succeed, amount: " + amount + ", businessKey:" + businessKey);
       return true;
    }
      @Override
      public boolean compensateReduce(String businessKey, Map<String, Object> params) {
            if(params != null && "true".equals(params.get("throwException"))){
                  throw new RuntimeException("compensate reduce balance failed");
            }
            LOGGER.info("compensate reduce balance succeed, businessKey:" + businessKey);
            return true;
      }
}
// 接?略
public class InventoryActionImpl implements InventoryAction {
     private static final Logger LOGGER = LoggerFactory.getLogger(InventoryActionImpl.class);
     @Override
     public boolean reduce(String businessKey, int count) {
          LOGGER.info("reduce inventory succeed, count: " + count + ", businessKey:" + businessKey);
          return true;
     }
     @Override
     public boolean compensateReduce(String businessKey) {
          LOGGER.info("compensate reduce inventory succeed, businessKey:" + businessKey);
          return true;
     }
}
复制代码

Seata XA模式

官??档

seata.io/zh-cn/docs/…

代码演示

添加Seata的依赖 & 配置
你平时的JDBC代码怎么写，依然怎么写。

4种模式对?与选择

AT

优势：

使?简单：对业务侵?性?

缺点：

性能中等
有全局锁

适?场景：

适?于对性能没有特别?的要求的场景
适?于不希望对业务进?改造的场景

TCC

优势

性能会? AT 模式?很多

缺点

相对于 AT 模式，TCC 模式对业务代码有?定的侵?性

适?场景：适?于核?系统等对性能有很?要求的场景

Saga

优势：

?阶段提交本地数据库事务，?锁，?性能；
参与者可以采?事务驱动异步执?，?吞吐；
补偿服务即正向服务的“反向”，易于理解，易于实现；

缺点：

Saga 模式由于?阶段已经提交本地数据库事务，且没有进?“预留”动作，所以不能保证隔离性。后续会讲到对于缺乏隔离性的应对措施。

适?场景：

业务流程?/多
参与者包含其他公司或遗留系统服务，?法提供 TCC 模式要求的三个接?
典型业务系统：如?融?络（与外部?融机构对接）、互联?微贷、渠道整合、分布式架构服务
集成等业务系统
银?业?融机构使??泛

XA

优势：

?侵?

缺点：

性能较差
需要数据库?持XA

适?场景：

强?致性的解决?案，适?于对?致性要求?常?的场景（使?较少）

【分布式技术专题】「架构实践于案例分析」总结和盘点目前常用分布式事务特别及问题分析（Seata-终）

分布式事务中间件对?与选择

Seata实现分布式事务

搭建Seata Server

Linux/Unix/Mac

Windows

启动时，也可指定参数

?持的参数如下表所示

Seata AT模式

官??档

代码演示

Maven依赖

配置

配置说明

创建Seata事务记录表

Controller代码：

Service代码：

被调??代码

Maven依赖

配置

Controller代码：

Service代码：

Seata TCC模式

需要实现的3个?法：

官??档

代码演示

行为操作接口

接口实现类

定义状态机?件：

Seata saga模式

方案简介

处理流程

Saga事务基本协议

代码实现

Seata XA模式

官??档

代码演示

4种模式对?与选择

AT

优势：

缺点：

适?场景：

TCC

优势

缺点

Saga

优势：

缺点：

适?场景：

XA

优势：

缺点：

适?场景：

热门文章

最新文章

相关课程

相关电子书

相关实验场景