Seata学习整理

路行的亚洲

发布于 2023-02-28 13:29:51

2680

发布于 2023-02-28 13:29:51

以下代码基于seata和seata-example

一、Seata使用的业务场景

在配车的业务中，我们使用了Seata的分布式事务来保证配车的业务逻辑能够正常时，才会做订单信息推送到财务系统。我们的系统配车业务一开始使用Seata的TCC模式来实现的，需要自己实现try和confirm或者cancel的逻辑。之后，由于seata推出了AT模式，之后系统采用的分布式事务使用的是AT模式。

相比如之前的TCC模式，AT模式只需要添加 @GlobalTransactional就可以实现分布式事务。

   @GlobalTransactional(rollbackFor = Exception.class)
    @Override
    public Integer confirmCarSubmit(MatchConfirmCarDTO matchConfirmCarDTO, LoginInfoDto loginInfoDto) {
        return retailOrderService.matchConfirmCarSubmit(null, matchConfirmCarDTO, loginInfoDto);
    }

配车的过程中，如果配了车，则需要修改库存中车辆库存状态为已锁定,配车状态为已配车,且释放原车。同时做财务信息推送。由于涉及到库存和财务信息，因此需要用到分布式事务。

那么为啥加了 @GlobalTransactional，它就可以实现分布式事务呢？

首先Seata分为两端，Seata Server和Seata Client。TC作为Seata的Server端，而RM和TM作为客户端。由于其是注解，因此，我们可以想象得到应该是基于全局事务注解。

下面我们下载seata的源码，基于seata的源码进行学习。

二、Seata服务端启动

首先启动Seata Server：

可以看到Seata Server主要做了这样几件事：

1）初始化监控度量信息

2）初始化改良版雪花算法UUID

3) 初始化SessionHolder

4) 初始化协调器TC,并将其设置为handler

5) 初始化netty服务端

    public static void main(String[] args) throws IOException {

        //initialize the metrics
        LOGGER.info("------初始化监控信息-----");
        MetricsManager.get().init();

        LOGGER.info("------初始化改良版雪花算法UUID-----");
        UUIDGenerator.init(parameterParser.getServerNode());

        LOGGER.info("------初始化SessionHolder-----");
        SessionHolder.init(parameterParser.getStoreMode());

        // 事务协调器，相当于netty的handler角色
        DefaultCoordinator coordinator = new DefaultCoordinator(nettyRemotingServer);
        LOGGER.info("------初始化协调器TC-----");
        coordinator.init();
        nettyRemotingServer.setHandler(coordinator);

        LOGGER.info("------初始化netty服务端-----");
        nettyRemotingServer.init();
    }

其中最重点的是coordinator.init(); 事务协调器初始化。

其主要启动了5个定时任务：

 handleRetryRollbacking();
 handleRetryCommitting();
 handleAsyncCommitting();
 timeoutCheck();
 undoLogDelete();

其中：

handleRetryRollbacking(); 处理重试回滚，每秒1次 handleRetryCommitting(); 处理重试提交，每秒1次 handleAsyncCommitting(); 处理异步提交，每秒1次 timeoutCheck(); 超时提交，每秒1次 undoLogDelete(); undo log删除，每3小时1次

三、Seata客户端启动

客户端启动的时候，可以看到其会执行GlobalTransactionScanner继承了InitializingBean和AbstractAutoProxyCreator。因此

@Override
    public void afterPropertiesSet() {
        ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
            (ConfigurationChangeListener)this);
        if (disableGlobalTransaction) {
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Global transaction is disabled.");
            }
            return;
        }
        if (initialized.compareAndSet(false, true)) {
            LOGGER.info("初始化客户端");
            initClient();
        }
    }

io.seata.spring.annotation.GlobalTransactionScanner#initClient方法，初始化客户端：

 private void initClient() {
        //init TM
        LOGGER.info("-------TM客户端初始化-------");
        TMClient.init(applicationId, txServiceGroup, accessKey, secretKey);

        //init RM
        LOGGER.info("-------RM客户端初始化-------");
        RMClient.init(applicationId, txServiceGroup);

        // 注册钩子方法
        registerSpringShutdownHook();

    }

主要做了三件事：

初始化TM，注册相关处理器，同时放入processorTable
初始化RM，注册相关处理器，同时放入processorTable
注册钩子方法

接着注册服务到seata中，然后Netty会执行channelRead执行事件处理：

 protected void processMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        Object body = rpcMessage.getBody();
        if (body instanceof MessageTypeAware) {
            // 根据messageType获取对应的处理器和线程池，processorTable 在 netty.init 时会初始化
            MessageTypeAware messageTypeAware = (MessageTypeAware) body;
            final Pair<RemotingProcessor, ExecutorService> pair = this.processorTable.get((int) messageTypeAware.getTypeCode());
            // 判断该Pair是否有初始化线程池，如果有就用业务线程池执行，否则直接执行
            if (pair != null) {
                if (pair.getSecond() != null) {
                    try {
                        //执行处理
                        pair.getSecond().execute(() -> {
                            try {
                                LOGGER.info("---执行处理processMessage-----");
                                pair.getFirst().process(ctx, rpcMessage);
                            } catch (Throwable th) {
                                LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
                            }
                        });
                    } catch (RejectedExecutionException e) {
                        LOGGER.error(FrameworkErrorCode.ThreadPoolFull.getErrCode(),
                            "thread pool is full, current max pool size is " + messageExecutor.getActiveCount());
                        if (allowDumpStack) {
                            String name = ManagementFactory.getRuntimeMXBean().getName();
                            String pid = name.split("@")[0];
                            int idx = new Random().nextInt(100);
                            try {
                                Runtime.getRuntime().exec("jstack " + pid + " >d:/" + idx + ".log");
                            } catch (IOException exx) {
                                LOGGER.error(exx.getMessage());
                            }
                            allowDumpStack = false;
                        }
                    }
                } else {
                    try {
                        pair.getFirst().process(ctx, rpcMessage);
                    } catch (Throwable th) {
                        LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
                    }
                }
            } else {
                LOGGER.error("This message type [{}] has no processor.", messageTypeAware.getTypeCode());
            }
        } else {
            LOGGER.error("This rpcMessage body[{}] is not MessageTypeAware type.", body);
        }
    }

判断该Pair是否有初始化线程池，如果有就用业务线程池执行，否则直接执行。

pair.getFirst().process(ctx, rpcMessage);

进入RM注册操作，通过Netty的RegRmProcessor。

RM register success,message:RegisterRMRequest{resourceIds='jdbc:mysql://127.0.0.1:3306/seata', applicationId='order-service',transactionServiceGroup='my_test_tx_group'},channel:[id: 0xb806e7c6, L:/127.0.0.1:8091 - R:/127.0.0.1:57663],client version:1.4.2

可以看到rm注册的信息：数据源连接信息、应用id、事务服务组、netty通道信息

进入Tm注册操作，通过Netty的RegTmProcessor。

TM register success,message:RegisterTMRequest{applicationId='account-service', transactionServiceGroup='my_test_tx_group'},channel:[id: 0xf2eafbcb, L:/127.0.0.1:8091 - R:/127.0.0.1:57691],client version:1.4.2

可以看到tm注册的信息：应用id、事务服务组、netty通道信息

四、执行业务系统业务

 /**
     * 购买下单，模拟全局事务提交
     *
     * @return
     */
    @RequestMapping("/purchase/commit")
    public Boolean purchaseCommit(HttpServletRequest request) {
        businessService.purchase("1001", "2001", 1);
        return true;
    }

  /**
     * 减库存，下订单
     *
     * @param userId
     * @param commodityCode
     * @param orderCount
     */
    @GlobalTransactional
    public void purchase(String userId, String commodityCode, int orderCount) {
        LOGGER.info("purchase begin ... xid: " + RootContext.getXID());
        stockClient.deduct(commodityCode, orderCount);
        orderClient.create(userId, commodityCode, orderCount);
    }

可以看到减库存，下订单信息：

此时我们可以看到有一个方法：io.seata.spring.annotation.GlobalTransactionalInterceptor#invoke

可以看到拦截器里面有一个invoke方法，此方法会获取全局事务注解和全局锁注解。

   final GlobalTransactional globalTransactionalAnnotation =getAnnotation(method, targetClass, GlobalTransactional.class);
   final GlobalLock globalLockAnnotation = getAnnotation(method, targetClass, GlobalLock.class);

根据对应的注解执行对应的处理：

handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);
handleGlobalLock(methodInvocation, globalLockAnnotation)

可以看到控制台打印的日志：

业务系统客户端：

发起的事务：

分支事务：

可以看到分支事务二阶段提交和提交状态committed。

可以看到这个过程经历了：

开启全局事务
创建全局session
session开启
channelRead执行业务处理processMessage
注册分支事务
提交全局事务
二阶段分支提交
二阶段提交发送
channelRead执行业务处理processMessage
执行处理processMessage
二阶段提交，删除分支
提交全局事务成功

其中：

在二阶段提交前，会先生成前镜像，然后执行业务sql，然后生成后镜像，准备事务日志。

完成后，会执行二阶段的提交操作。