分布式事务saga实现的理论基础Hector&Kenneth在1987年发表的论文Sagas，它的核心思想是当整个事务的一个节点失败后，依赖于状态对当前事务从前到后进行重试，或者从后往前进行补偿。
saga模式的主要应用场景是业务流程比较长，有一些服务不能提供TCC模式的三个接口，或者不能实现AT模式的依赖undolog实现自动补偿。
阿里的seata中间件是通过状态机来实现的，它使用状态图定义服务调用流程并生成json状态语言定义文件，状态图的节点可以是一个服务，也可以是补偿节点。这个生成的json由状态机引擎来驱动执行，出现异常是状态机引擎对调用成功的服务从后往前补偿，而补偿的逻辑需要由服务自己来实现。
本文我们还是用之前TCC模式中的例子，我们在电商网站购买一件商品，后台首先会从订单服务下单，然后订单服务会调用账户服务扣减商品金额，如果成功，再调用库存服务扣减库存。如果其中某一步失败，则从后往前依次补偿，这个补偿事件由状态机触发。
配置状态机
首先我们需要创建3张表，sql语句如下，注意，本文使用的mysql：
create table seata_state_machine_def ( id varchar(32) not null comment 'id', name varchar(128) not null comment 'name', tenant_id varchar(32) not null comment 'tenant id', app_name varchar(32) not null comment 'application name', type varchar(20) comment 'state language type', comment_ varchar(255) comment 'comment', ver varchar(16) not null comment 'version', gmt_create timestamp(3) not null comment 'create time', status varchar(2) not null comment 'status(AC:active|IN:inactive)', content longtext comment 'content', recover_strategy varchar(16) comment 'transaction recover strategy(compensate|retry)', primary key (id) ); CREATE TABLE seata_state_machine_inst ( id VARCHAR(128) NOT NULL COMMENT 'id', machine_id VARCHAR(32) NOT NULL COMMENT 'state machine definition id', tenant_id VARCHAR(32) NOT NULL COMMENT 'tenant id', parent_id VARCHAR(128) COMMENT 'parent id', gmt_started TIMESTAMP(3) NOT NULL COMMENT 'start time', business_key VARCHAR(48) COMMENT 'business key', start_params LONGTEXT COMMENT 'start parameters', gmt_end TIMESTAMP(3) COMMENT 'end time', excep BLOB COMMENT 'exception', end_params LONGTEXT COMMENT 'end parameters', STATUS VARCHAR(2) COMMENT 'status(SU succeed|FA failed|UN unknown|SK skipped|RU running)', compensation_status VARCHAR(2) COMMENT 'compensation status(SU succeed|FA failed|UN unknown|SK skipped|RU running)', is_running TINYINT(1) COMMENT 'is running(0 no|1 yes)', gmt_updated TIMESTAMP(3) NOT NULL, PRIMARY KEY (id), UNIQUE KEY unikey_buz_tenant (business_key, tenant_id) ); CREATE TABLE seata_state_inst ( id VARCHAR(48) NOT NULL COMMENT 'id', machine_inst_id VARCHAR(128) NOT NULL COMMENT 'state machine instance id', NAME VARCHAR(128) NOT NULL COMMENT 'state name', TYPE VARCHAR(20) COMMENT 'state type', service_name VARCHAR(128) COMMENT 'service name', service_method VARCHAR(128) COMMENT 'method name', service_type VARCHAR(16) COMMENT 'service type', business_key VARCHAR(48) COMMENT 'business key', state_id_compensated_for VARCHAR(50) COMMENT 'state compensated for', state_id_retried_for VARCHAR(50) COMMENT 'state retried for', gmt_started TIMESTAMP(3) NOT NULL COMMENT 'start time', is_for_update TINYINT(1) COMMENT 'is service for update', input_params LONGTEXT COMMENT 'input parameters', output_params LONGTEXT COMMENT 'output parameters', STATUS VARCHAR(2) NOT NULL COMMENT 'status(SU succeed|FA failed|UN unknown|SK skipped|RU running)', excep BLOB COMMENT 'exception', gmt_end TIMESTAMP(3) COMMENT 'end time', PRIMARY KEY (id, machine_inst_id) );
本文电商购物的流程状态，我们用下图表示：
注意：seata提供了下面地址可以绘制这个图，同时生成对应的json代码。本文的json代码是参考官方示例手工改写的。
http://seata.io/saga_designer/index.html#/
状态机需要依赖一个json，这个json定义了上面流程图中的节点，代码如下：
{
"Name": "buyGoodsOnline",
"Comment": "buy a goods on line, add order, deduct account, deduct storage ",
"StartState": "SaveOrder",
"Version": "0.0.1",
"States": {
"SaveOrder": {
"Type": "ServiceTask",
"ServiceName": "orderSave",
"ServiceMethod": "saveOrder",
"CompensateState": "DeleteOrder",
"Next": "ChoiceAccountState",
"Input": [
"$.[businessKey]",
"$.[order]"
],
"Output": {
"SaveOrderResult": "$.#root"
},
"Status": {
"#root == true": "SU",
"#root == false": "FA",
"$Exception{java.lang.Throwable}": "UN"
}
},
"ChoiceAccountState":{
"Type": "Choice",
"Choices":[
{
"Expression":"[SaveOrderResult] == true",
"Next":"ReduceAccount"
}
],
"Default":"Fail"
},
"ReduceAccount": {
"Type": "ServiceTask",
"ServiceName": "accountService",
"ServiceMethod": "decrease",
"CompensateState": "CompensateReduceAccount",
"Next": "ChoiceStorageState",
"Input": [
"$.[businessKey]",
"$.[userId]",
"$.[money]",
{
"throwException" : "$.[mockReduceAccountFail]"
}
],
"Output": {
"ReduceAccountResult": "$.#root"
},
"Status": {
"#root == true": "SU",
"#root == false": "FA",
"$Exception{java.lang.Throwable}": "UN"
},
"Catch": [
{
"Exceptions": [
"java.lang.Throwable"
],
"Next": "CompensationTrigger"
}
]
},
"ChoiceStorageState":{
"Type": "Choice",
"Choices":[
{
"Expression":"[ReduceAccountResult] == true",
"Next":"ReduceStorage"
}
],
"Default":"Fail"
},
"ReduceStorage": {
"Type": "ServiceTask",
"ServiceName": "storageService",
"ServiceMethod": "decrease",
"CompensateState": "CompensateReduceStorage",
"Input": [
"$.[businessKey]",
"$.[productId]",
"$.[count]",
{
"throwException" : "$.[mockReduceStorageFail]"
}
],
"Output": {
"ReduceStorageResult": "$.#root"
},
"Status": {
"#root == true": "SU",
"#root == false": "FA",
"$Exception{java.lang.Throwable}": "UN"
},
"Catch": [
{
"Exceptions": [
"java.lang.Throwable"
],
"Next": "CompensationTrigger"
}
],
"Next": "Succeed"
},
"DeleteOrder": {
"Type": "ServiceTask",
"ServiceName": "orderSave",
"ServiceMethod": "deleteOrder",
"Input": [
"$.[businessKey]",
"$.[order]"
]
},
"CompensateReduceAccount": {
"Type": "ServiceTask",
"ServiceName": "accountService",
"ServiceMethod": "compensateDecrease",
"Input": [
"$.[businessKey]",
"$.[userId]",
"$.[money]"
]
},
"CompensateReduceStorage": {
"Type": "ServiceTask",
"ServiceName": "storageService",
"ServiceMethod": "compensateDecrease",
"Input": [
"$.[businessKey]",
"$.[productId]",
"$.[count]"
]
},
"CompensationTrigger": {
"Type": "CompensationTrigger",
"Next": "Fail"
},
"Succeed": {
"Type":"Succeed"
},
"Fail": {
"Type":"Fail",
"ErrorCode": "PURCHASE_FAILED",
"Message": "purchase failed"
}
}
}
上面的json中，我们定义了6个ServiceTask,分别对应订单服务保存订单、账户服务扣减账户和库存服务扣减库存以及对应的补偿机制。
我们定义了CompensationTrigger，并且在账户服务和库存服务抛出异常后，会调用CompensationTrigger来触发补偿事件。
对于每一个节点，我们需要定义type，同时对于ServiceTask类型的节点，我们需要定义触发方法，回滚事件对应的ServiceTask，下一个流程节点，输入/输出参数、异常等。
订单服务
订单服务是本文中的集成服务，它会调用账户服务和库存服务来实现业务。
首先，前端发起下单请求后，订单服务会接收这个服务，然后启动状态机，这个代码在OrderServiceImpl，代码如下：
public boolean create(Order order) {
LOGGER.info("------->交易开始");
StateMachineEngine stateMachineEngine = (StateMachineEngine) ApplicationContextUtils.getApplicationContext().getBean("stateMachineEngine");
Map<String, Object> startParams = new HashMap<>(3);
String businessKey = String.valueOf(System.currentTimeMillis());
startParams.put("businessKey", businessKey);
startParams.put("order", order);
startParams.put("mockReduceAccountFail", "true");
startParams.put("userId", order.getUserId());
startParams.put("money", order.getPayAmount());
startParams.put("productId", order.getProductId());
startParams.put("count", order.getCount());
//这里状态机是同步的，seata也支持异步，可以参考官方示例
StateMachineInstance inst = stateMachineEngine.startWithBusinessKey("buyGoodsOnline", null, businessKey, startParams);
Assert.isTrue(ExecutionStatus.SU.equals(inst.getStatus()), "saga transaction execute failed. XID: " + inst.getId());
System.out.println("saga transaction commit succeed. XID: " + inst.getId());
inst = stateMachineEngine.getStateMachineConfig().getStateLogStore().getStateMachineInstanceByBusinessKey(businessKey, null);
Assert.isTrue(ExecutionStatus.SU.equals(inst.getStatus()), "saga transaction execute failed. XID: " + inst.getId());
return true;
}
保存账单OrderSaveImpl对应json里面的orderSave，里面的回滚方法就是deleteOrder，代码如下：
public class OrderSaveImpl implements OrderApi{
private Logger logger = LoggerFactory.getLogger(getClass());
@Resource
private OrderDao orderDao;
@Override
public boolean saveOrder(String businessKey, Order order) {
logger.info("保存订单, businessKey：{}, order: {}", businessKey, order);
orderDao.create(order);
return true;
}
/**
* 回滚事务,删除订单
* @param order order
* @return
*/
@Override
public boolean deleteOrder(String businessKey,Order order){
logger.info("删除订单, businessKey：{}, order: {}", businessKey, order);
orderDao.delete(order);
return true;
}
}
定义调用账户服务，这里使用feign调用，补偿方法是compensateDecrease，对应json文件中的accountService，代码如下：
public class AccountServiceImpl implements AccountService{
@Resource
private AccountApi accountApi;
@Override
public boolean decrease(String businessKey, Long userId, BigDecimal money) {
return accountApi.decrease(businessKey, userId, money);
}
@Override
public boolean compensateDecrease(String businessKey, Long userId, BigDecimal money) {
return accountApi.compensateDecrease(businessKey, userId, money);
}
}
定义调用库存服务，这里使用feign调用，补偿方法是compensateDecrease，对应json文件中的storageService，代码如下：
public class StorageServiceImpl implements StorageService{
@Resource
private StorageApi storageApi;
@Override
public boolean decrease(String businessKey, Long productId, Integer count) {
return storageApi.decrease(businessKey, productId, count);
}
@Override
public boolean compensateDecrease(String businessKey, Long productId, Integer count) {
return storageApi.compensateDecrease(businessKey, productId, count);
}
}
库存服务
看了订单服务代码，再看库存服务就非常简单了，为订单服务提供http接口，包括2个方法，扣减库存和补偿扣减库存，controller调用的service代码如下：
@Service("storageService")
public class StorageServiceImpl implements StorageService {
private static final Logger LOGGER = LoggerFactory.getLogger(StorageServiceImpl.class);
@Resource
private StorageDao storageDao;
@Override
public boolean decrease(Long productId, Integer count) {
LOGGER.info("扣减库存, commit, productId:{}, count:{}", productId, count);
storageDao.decrease(productId, count);
//throw new RuntimeException();
}
@Override
public boolean compensateDecrease(Long productId, Integer count) {
LOGGER.info("补偿扣减库存, compensate, productId:{}, count:{}", productId, count);
storageDao.compensateDecrease(productId, count);
return true;
}
}
账户服务的代码跟这个类似，就不再贴代码了。
测试
本文的实验中，订单、账户、库存服务都有自己的数据库，这里不再贴sql了，需要了解的可以看我之前的文章《springcloud+eureka整合seata-AT模式》，或者下载源代码，文末有源码地址。
开始实验之前，订单表没有数据，账户和库存数据如下：
account表：
storage表：
我们进行一次成功的实验，向下面的url发送购买商品请求：
http://localhost:8180/order/create
{
"userId":1,
"productId":1,
"count":1,
"money":1,
"payAmount":50
}
成功之后发现订单表有了数据，账户表和库存表数据如下：
order表：
account表
storage表
我们把库存服务的decrease方法改成如下：
public boolean decrease(Long productId, Integer count) {
LOGGER.info("扣减库存, commit, productId:{}, count:{}", productId, count);
storageDao.decrease(productId, count);
throw new RuntimeException();
}
这时发送购买商品请求后会抛出异常，然后3个服务的事务依次做交易补偿，所有表数据没有变。
总结
seata中的saga模式适用于长流程或者长事务的场景。而saga模式复杂的地方在于引入了状态机，需要定义状态机的流程，把定义好的流程用json文件引入工程中。
同时saga模式需要开发者自己定义回滚事件，如果回滚失败，对整个事务的控制就非常复杂了。
本文源码地址：
https://github.com/jinjunzhu/springcloud-eureka-seata-saga.git