阿里开源分布式事务组件 seata ：demo 环境搭建以及运行流程简析
案例设计
seata 官方给出了一系列 demo 样例，不过我在用的过程中发现总有这个那个的问题，所以自己维护了一份基于 dubbo 的 demo 在 github 上，适配的 seata 版本是 0.8.0。
案例的设计直接参考官方 quick start给出的案例：
整个案例分为三个服务，分别是存储服务、订单服务和账户服务，这些服务通过 dubbo 进行发布和调用，内部调用逻辑如上面图所示。
整个 demo 的工程样例如下所示：
undo_log 表
这个案例除了在数据库需要建立业务表以外，还要额外建立一张 undo_log 表，这个表的主要作用是记录事务的前置镜像和后置镜像。
全局事务进行到提交阶段，则删除该表对应的记录，全局事务如果需要回滚，则会利用这个表里记录的镜像数据，恢复数据。
undo_log 表里的数据实际上是“朝生夕死”的，数据不需要在表里存活太久。表结构如下所示：
CREATE TABLE `undo_log` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`branch_id` bigint(20) NOT NULL,
`xid` varchar(100) NOT NULL,
`context` varchar(128) NOT NULL,
`rollback_info` longblob NOT NULL,
`log_status` int(11) NOT NULL,
`log_created` datetime NOT NULL,
`log_modified` datetime NOT NULL,
`ext` varchar(100) DEFAULT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
服务逻辑
每个服务都对应了一个 starter 类，这个类主要用来在 spring 环境下，将该服务启动，并通过 dubbo 发布出去，以账户服务为例：
/**
* The type Dubbo account service starter.
*/
public class DubboAccountServiceStarter {
/**
* 2. Account service is ready . A buyer register an account: U100001 on my e-commerce platform
*
* @param args the input arguments
*/
public static void main(String[] args) {
ClassPathXmlApplicationContext accountContext = new ClassPathXmlApplicationContext(new String[]{"spring/dubbo-account-service.xml"});
accountContext.getBean("service");
JdbcTemplate accountJdbcTemplate = (JdbcTemplate) accountContext.getBean("jdbcTemplate");
accountJdbcTemplate.update("delete from account_tbl where user_id = 'U100001'");
accountJdbcTemplate.update("insert into account_tbl(user_id, money) values ('U100001', 999)");
new ApplicationKeeper(accountContext).keep();
}
}
首先通过
ClassPathXmlApplicationContext
读取 dubbo-account-service.xml 这个 spring 配置文件并启动 spring 容器环境，并通过 spring 的 jdbc template 对账户表的数据进行初始化。
dubbo-account-service.xml 配置文件中进行了各类 bean 的配置，包括 dubbo 与 spring 结合时的标准配置：
<bean id="accountDataSourceProxy" class="io.seata.rm.datasource.DataSourceProxy">
<constructor-arg ref="accountDataSource" />
</bean>
<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
<property name="dataSource" ref="accountDataSourceProxy" />
</bean>
<dubbo:application name="dubbo-demo-account-service" />
<dubbo:registry address="zookeeper://localhost:2181" />
<dubbo:protocol name="dubbo" port="20881" />
<dubbo:service interface="io.seata.samples.dubbo.service.AccountService" ref="service" timeout="10000"/>
<bean id="service" class="io.seata.samples.dubbo.service.impl.AccountServiceImpl">
<property name="jdbcTemplate" ref="jdbcTemplate"/>
</bean>
<bean class="io.seata.spring.annotation.GlobalTransactionScanner">
<constructor-arg value="dubbo-demo-account-service"/>
<constructor-arg value="my_test_tx_group"/>
</bean>
这份配置里主要有两个需要引起注意的关键点
-
jdbcTemplate 这个 bean 所依赖的数据源 bean，是一个类名为 io.seata.rm.datasource.DataSourceProxy 的数据源类，通过它的名字可以很明显地看出这是一个代理模式的应用，因为 seata 为完成全局事务的逻辑，需要在普通的 sql 操作前后添加一些逻辑，比如说 sql 执行前对 sql 进行语法解析，生成前置镜像，sql 执行后生成后置镜像，通过代理的方式，可以方便地对 connection，statement 等进行代理包装，在调用的时候进行拦截，加入自己的逻辑。
-
配置文件中还有一个 io.seata.spring.annotation.GlobalTransactionScanner 类型的 bean，这个 bean 是支撑 seata 能在 spring 环境中通过注解的方式来划定事务边界的基础。在 spring 容器启动时，会扫描
@GlobalTransactional注解是否存在，这个注解标识了全局事务的开始和结束，也就是我们常说的“事务的边界”
业务逻辑
业务逻辑的具体详情在
BusinessServiceImpl
类中可以看到：
@Override
@GlobalTransactional(timeoutMills = 300000, name = "dubbo-demo-tx")
public void purchase(String userId, String commodityCode, int orderCount) {
LOGGER.info("purchase begin ... xid: " + RootContext.getXID());
storageService.deduct(commodityCode, orderCount);
orderService.create(userId, commodityCode, orderCount);
// throw new RuntimeException("xxx");
}
先调用存储服务，减少库存，然后调用订单服务，新建订单。这两个动作属于一个整体的事务，任何一个动作失败，都需要撤销所有的操作。
这个方法也有两个需要注意的点：
-
该方法上声明了 @GlobalTransactional(timeoutMills = 300000, name = "dubbo-demo-tx") 这样的注解，用于让上文提到的 GlobalTransactionScanner 扫描的时候发现这是一个全局事务。
-
方法的最后有一行代码抛出了 RuntimeException，这主要是为了模仿全局事务的失败，并让 seata 走全局事务回滚逻辑。
事务扫描与边界定义
上文提到的 GlobalTransactionScanner 类，会在 spring 容器启动的时候，也被初始化。
在它的 afterPropertiesSet 方法被调用时，会触发 seata client 的初始化
@Override
public void afterPropertiesSet() {
if (disableGlobalTransaction) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Global transaction is disabled.");
}
return;
}
initClient();
}
关于 seata client 的初始化的细节，可以看看我写的另外一篇文章 《阿里开源分布式事务组件 seata ： seata client 通信层解析》 。
初始化客户端做的事情主要是建立与 seata server 的连接，并注册 TM 和 RM。接下来，在 wrapIfNecessary 方法里，实现对注解的扫描，并对添加了注解的方法添加 interceptor。
这篇文章里我们暂时不讨论 TCC 模式，只讨论 AT 模式，也暂不讨论全局事务锁 GlobalLock 的实现，先忽略这些有关的逻辑，只关注事务处理逻辑。
Class<?> serviceInterface = SpringProxyUtils.findTargetClass(bean);
Class<?>[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean);
if (!existsAnnotation(new Class[] {serviceInterface})
&& !existsAnnotation(interfacesIfJdk)) {
return bean;
}
if (interceptor == null) {
interceptor = new GlobalTransactionalInterceptor(failureHandlerHook);
}
在这里，interceptor 的实现是 GlobalTransactionalInterceptor，也就是说，以上文的案例为例子，当 BusinessServiceImpl 的 purchase 方法被调用的时候，实际上这个方法会被拦截器拦截，执行拦截器里的逻辑：
@Override
public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
Class<?> targetClass = (methodInvocation.getThis() != null ? AopUtils.getTargetClass(methodInvocation.getThis()) : null);
Method specificMethod = ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass);
final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
final GlobalTransactional globalTransactionalAnnotation = getAnnotation(method, GlobalTransactional.class);
final GlobalLock globalLockAnnotation = getAnnotation(method, GlobalLock.class);
if (globalTransactionalAnnotation != null) {
return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);
} else if (globalLockAnnotation != null) {
return handleGlobalLock(methodInvocation);
} else {
return methodInvocation.proceed();
}
}
private Object handleGlobalTransaction(final MethodInvocation methodInvocation,
final GlobalTransactional globalTrxAnno) throws Throwable {
try {
return transactionalTemplate.execute(new TransactionalExecutor() {
@Override
public Object execute() throws Throwable {
return methodInvocation.proceed();
}
public String name() {
String name = globalTrxAnno.name();
if (!StringUtils.isNullOrEmpty(name)) {
return name;
}
return formatMethod(methodInvocation.getMethod());
}
@Override
public TransactionInfo getTransactionInfo() {
TransactionInfo transactionInfo = new TransactionInfo();
transactionInfo.setTimeOut(globalTrxAnno.timeoutMills());
transactionInfo.setName(name());
Set<RollbackRule> rollbackRules = new LinkedHashSet<>();
for (Class<?> rbRule : globalTrxAnno.rollbackFor()) {
rollbackRules.add(new RollbackRule(rbRule));
}
for (String rbRule : globalTrxAnno.rollbackForClassName()) {
rollbackRules.add(new RollbackRule(rbRule));
}
for (Class<?> rbRule : globalTrxAnno.noRollbackFor()) {
rollbackRules.add(new NoRollbackRule(rbRule));
}
for (String rbRule : globalTrxAnno.noRollbackForClassName()) {
rollbackRules.add(new NoRollbackRule(rbRule));
}
transactionInfo.setRollbackRules(rollbackRules);
return transactionInfo;
}
});
} catch (TransactionalExecutor.ExecutionException e) {
TransactionalExecutor.Code code = e.getCode();
switch (code) {
case RollbackDone:
throw e.getOriginalException();
case BeginFailure:
failureHandler.onBeginFailure(e.getTransaction(), e.getCause());
throw e.getCause();
case CommitFailure:
failureHandler.onCommitFailure(e.getTransaction(), e.getCause());
throw e.getCause();
case RollbackFailure:
failureHandler.onRollbackFailure(e.getTransaction(), e.getCause());
throw e.getCause();
default:
throw new ShouldNeverHappenException("Unknown TransactionalExecutor.Code: " + code);
}
}
}
在执行 handleGlobalTransaction 方法时，实际上采用模板模式，委托给了 TransactionalTemplate 类去执行标准的事务处理流程。如下所示：
/**
* Execute object.
*
* @param business the business
* @return the object
* @throws TransactionalExecutor.ExecutionException the execution exception
*/
public Object execute(TransactionalExecutor business) throws Throwable {
// 1. get or create a transaction
GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
// 1.1 get transactionInfo
TransactionInfo txInfo = business.getTransactionInfo();
if (txInfo == null) {
throw new ShouldNeverHappenException("transactionInfo does not exist");
}
try {
// 2. begin transaction
beginTransaction(txInfo, tx);
Object rs = null;
try {
// Do Your Business
rs = business.execute();
} catch (Throwable ex) {
// 3.the needed business exception to rollback.
completeTransactionAfterThrowing(txInfo,tx,ex);
throw ex;
}
// 4. everything is fine, commit.
commitTransaction(tx);
return rs;
} finally {
//5. clear
triggerAfterCompletion();
cleanUp();
}
}
事务处理逻辑实际上是一种模板，将事务相关的处理逻辑放在 try 块里，发现异常后执行回滚，正常执行则执行提交。
在这里有个需要注意的地方是，seata 不把提交这个动作放在 try 块里，因为在 seata 里，全局事务的提交实际上是可以异步执行的。
因为全局事务如果进行到提交这一阶段，那么意味着各个分支事务已经执行过本地提交，全局事务的提交阶段仅仅是删除 undo_log 里的记录，这个记录删除或者不删除，实际上不会改变全局事务已经正常完成的事实。所以它可以用程序异步去做，或者以人工介入的方式去做，所以 seata 认为，全局事务提交失败，不需要执行回滚流程。
正文到此结束
- 本文标签： tab 数据库 NSA 分布式事务 dubbo ACE executor Proxy classpath find IDE 解析 开源 删除 ORM Statement root tar 数据 HTML setTimeout http bean JDBC update UI App final core Service bus spring https 配置 CTO 锁 src Action 分布式 git sql 代码 value client Property id struct zookeeper GitHub key Connection trigger AOP HashSet db cat API XML dataSource 文章 IO 对账
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
近期评论
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
搞个gitee的项目
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
