Seata解析

文章23 |   阅读 13235 |   点赞0

来源：https://blog.csdn.net/weixin_38308374/category_10261111.html

Seata解析-详解RM及TM的初始化过程

转x33g5p2x  于2021-12-21 转载在 其他  
字(10.2k)|赞(0)|评价(0)|浏览(638)

本文基于seata 1.3.0版本

虽然RM和TM功能不相同，但是两者都部署在应用端，和应用程序部署在一起，RM和TM的初始化都是由同一位置触发的。
本文将详细介绍RM和TM的初始化过程。

一、GlobalTransactionScanner对象的创建

RM、TM的初始化都是由类GlobalTransactionScanner触发的，那么先来介绍GlobalTransactionScanner对象是如何创建的。
创建GlobalTransactionScanner对象时，需要设置两个参数：

  1. spring.application.name：应用名
  2. spring.cloud.alibaba.seata.tx-service-group：事务分组

1、自己编程创建

我们可以通过自己编程创建GlobalTransactionScanner对象。代码如下：

@Bean
	public GlobalTransactionScanner globalTransactionScanner() {
		//应用名
		String applicationName = applicationContext.getEnvironment()
				.getProperty("spring.application.name");
		//事务分组
		String txServiceGroup = applicationContext.getEnvironment()
				.getProperty("spring.cloud.alibaba.seata.tx-service-group");

		return new GlobalTransactionScanner(applicationName, txServiceGroup);
	}

2、使用GlobalTransactionAutoConfiguration

GlobalTransactionAutoConfiguration是由spring-cloud-alibaba-seata提供的，需要在pom.xml文件中引入spring-cloud-alibaba-seata。
GlobalTransactionAutoConfiguration是一个自动配置类，因此在spring boot启动的时候便可以加载，并创建需要的bean。
其创建GlobalTransactionScanner的方法如下，与上一小节的方法代码类似：

@Bean
	public GlobalTransactionScanner globalTransactionScanner() {

		String applicationName = applicationContext.getEnvironment()
				.getProperty("spring.application.name");

		String txServiceGroup = seataProperties.getTxServiceGroup();
		//如果没有配置事务分组，
		//则使用应用名+-fescar-service-group作为事务分组
		if (StringUtils.isEmpty(txServiceGroup)) {
			txServiceGroup = applicationName + "-fescar-service-group";
			seataProperties.setTxServiceGroup(txServiceGroup);
		}
		return new GlobalTransactionScanner(applicationName, txServiceGroup);
	}

3、使用SeataAutoConfiguration

SeataAutoConfiguration是由seata-spring-boot-starter提供的，该类也是一个自动配置类。在类中提供了方法globalTransactionScanner()来创建GlobalTransactionScanner对象，该方法的代码与GlobalTransactionAutoConfiguration类似，不再介绍。

二、GlobalTransactionScanner初始化

GlobalTransactionScanner实现了InitializingBean接口，所以在GlobalTransactionScanner对象创建完毕之后，spring容器执行afterPropertiesSet()方法。
afterPropertiesSet方法主要是初始化TM和RM。

public void afterPropertiesSet() {
		//disableGlobalTransaction表示是否启用全局事务，
		//如果设置false，则seata的分布式事务不起作用
		//可以通过transport.enableClientBatchSendRequest设置
        if (disableGlobalTransaction) {
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Global transaction is disabled.");
            }
            return;
        }
        //初始化
        initClient();
    }
    private void initClient() {
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Initializing Global Transaction Clients ... ");
        }
        if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {
            throw new IllegalArgumentException(String.format("applicationId: %s, txServiceGroup: %s", applicationId, txServiceGroup));
        }
        //init TM
        //TM初始化
        //TM的初始化主要完成以下几件事：
        //1、创建连接池
        //2、创建客户端Netty，并启动
        //3、创建并启动用于检测的线程池
        TMClient.init(applicationId, txServiceGroup);
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Transaction Manager Client is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
        }
        //init RM
        //RM初始化，其初始化过程与TM类似
        RMClient.init(applicationId, txServiceGroup);
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Resource Manager is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
        }

        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Global Transaction Clients are initialized. ");
        }
        //注册关闭回调钩子
        registerSpringShutdownHook();

    }

从initClient方法中可以看到，其主要做了TM和RM的初始化。RM的初始化与TM的初始化动作类似，下面重点分析TM。

1、TM初始化

TM的初始化主要完成以下三件事：

  1. 创建连接池
  2. 创建客户端Netty，并启动
  3. 创建并启动用于检测的线程池

下面从代码上来分析一下如何实现的。
TMClient的init方法如下：

public static void init(String applicationId, String transactionServiceGroup) {
        TmNettyRemotingClient tmNettyRemotingClient = TmNettyRemotingClient.getInstance(applicationId, transactionServiceGroup);
        tmNettyRemotingClient.init();
    }

init()方法首先获取TmNettyRemotingClient实例，然后调用其init()方法。下面来看一下TmNettyRemotingClient.getInstance()方法。

public static TmNettyRemotingClient getInstance(String applicationId, String transactionServiceGroup) {
        TmNettyRemotingClient tmNettyRemotingClient = getInstance();
        //设置应用名
        tmNettyRemotingClient.setApplicationId(applicationId);
        //设置服务分组 
        tmNettyRemotingClient.setTransactionServiceGroup(transactionServiceGroup);
        return tmNettyRemotingClient;
    }
    public static TmNettyRemotingClient getInstance() {
        if (instance == null) {
            synchronized (TmNettyRemotingClient.class) {
                if (instance == null) {
                    //Netty配置对象，针对Netty的大部分配置信息都在该类中
                    NettyClientConfig nettyClientConfig = new NettyClientConfig();
                    final ThreadPoolExecutor messageExecutor = new ThreadPoolExecutor(
                        nettyClientConfig.getClientWorkerThreads(), nettyClientConfig.getClientWorkerThreads(),
                        KEEP_ALIVE_TIME, TimeUnit.SECONDS,
                        new LinkedBlockingQueue<>(MAX_QUEUE_SIZE),
                        new NamedThreadFactory(nettyClientConfig.getTmDispatchThreadPrefix(),
                            nettyClientConfig.getClientWorkerThreads()),
                        RejectedPolicies.runsOldestTaskPolicy());
                    //创建TmNettyRemotingClient对象
                    instance = new TmNettyRemotingClient(nettyClientConfig, null, messageExecutor);
                }
            }
        }
        return instance;
    }
    //TmNettyRemotingClient父类AbstractNettyRemotingClient的构造方法如下：
    public AbstractNettyRemotingClient(NettyClientConfig nettyClientConfig, EventExecutorGroup eventExecutorGroup,
                                       ThreadPoolExecutor messageExecutor, NettyPoolKey.TransactionRole transactionRole) {
        super(messageExecutor);
        this.transactionRole = transactionRole;
        //NettyClientBootstrap的作用是管理Netty客户端，启动Netty、关闭Netty、创建新的链接
        clientBootstrap = new NettyClientBootstrap(nettyClientConfig, eventExecutorGroup, transactionRole);
        //设置Netty的责任链
        clientBootstrap.setChannelHandlers(new ClientHandler());
        //clientChannelManager是一个连接池管理器，管理与服务端的连接，当需要连接时，直接从连接池中获取
        //如果连接在连接池中不存在，则通过clientBootstrap获得新连接
        clientChannelManager = new NettyClientChannelManager(
            new NettyPoolableFactory(this, clientBootstrap), getPoolKeyFunction(), nettyClientConfig);
    }

TmNettyRemotingClient对象实例化时，主要是调用父类的构造方法，也就是AbstractNettyRemotingClient的构造方法。在AbstractNettyRemotingClient构造方法中，创建了Netty客户端启动器，设置了Netty的处理器，创建了连接池管理器。这里都只是创建了对象，比如Netty客户端启动器只是创建了对象，并没有启动Netty客户端，连接池管理器也是一样，连接池管理器中并没有连接。
到这里TmNettyRemotingClient实例创建完毕，下面来看一下TmNettyRemotingClient的init方法。

public void init() {
        //注册处理器
        registerProcessor();
        if (initialized.compareAndSet(false, true)) {
            //启动客户端Netty，启动检查连接是否可用的定时任务
            //代码见下方
            super.init();
        }
    }
    private void registerProcessor() {
        // 1.registry TC response processor
        //注册处理器
        //除了心跳报文外，其他的消息都是由ClientOnResponseProcessor处理的
        //ClientOnResponseProcessor用于处理响应消息，Netty会将响应消息交给本处理器处理
        ClientOnResponseProcessor onResponseProcessor =
            new ClientOnResponseProcessor(mergeMsgMap, super.getFutures(), getTransactionMessageHandler());
        super.registerProcessor(MessageType.TYPE_SEATA_MERGE_RESULT, onResponseProcessor, null);
        super.registerProcessor(MessageType.TYPE_GLOBAL_BEGIN_RESULT, onResponseProcessor, null);
        super.registerProcessor(MessageType.TYPE_GLOBAL_COMMIT_RESULT, onResponseProcessor, null);
        super.registerProcessor(MessageType.TYPE_GLOBAL_REPORT_RESULT, onResponseProcessor, null);
        super.registerProcessor(MessageType.TYPE_GLOBAL_ROLLBACK_RESULT, onResponseProcessor, null);
        super.registerProcessor(MessageType.TYPE_GLOBAL_STATUS_RESULT, onResponseProcessor, null);
        super.registerProcessor(MessageType.TYPE_REG_CLT_RESULT, onResponseProcessor, null);
        // 2.registry heartbeat message processor
        ClientHeartbeatProcessor clientHeartbeatProcessor = new ClientHeartbeatProcessor();
        super.registerProcessor(MessageType.TYPE_HEARTBEAT_MSG, clientHeartbeatProcessor, null);
    }
    //下面代码是TmNettyRemotingClient父类AbstractNettyRemotingClient的init()方法
    public void init() {
        //启动检测连接是否可用的定时任务
        //每过10s定时任务启动一次检查与TC的连接是否可用，如果发现不可用，则自动重连
        timerExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                clientChannelManager.reconnect(getTransactionServiceGroup());
            }
        }, SCHEDULE_DELAY_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);
        //判断是否允许客户端批量发送请求消息，默认是true
        if (NettyClientConfig.isEnableClientBatchSendRequest()) {
            //如果允许，则启动一个线程池，该线程池循环检查是否有可发送的消息，如果有，则收集所有发送到同一机器的消息，
            //然后将这些消息打包成一个对象MergedWarpMessage，之后将该对象发送到服务端。
            mergeSendExecutorService = new ThreadPoolExecutor(MAX_MERGE_SEND_THREAD,
                MAX_MERGE_SEND_THREAD,
                KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));
            mergeSendExecutorService.submit(new MergedSendRunnable());
        }
        //调用父类的init方法，在父类方法中启动了一个定时任务，用于清理超时请求
        //客户端发送每个请求都会记录到集合futures中，等待服务端返回后再从集合中删除
        //如果服务端超时，只能通过定时任务删除了
        super.init();
        //启动客户端Netty，下面方法执行完成，表示客户端可以建立与服务端的连接了
        clientBootstrap.start();
    }

TmNettyRemotingClient的init方法主要完成下面四项工作：

  1. 启动检测连接是否可用的定时任务；
  2. 启动批量发送请求消息的线程池；
  3. 启动定时任务，清理超时请求
  4. 启动客户端Netty，将ClientHandler设置到责任链中。

2、RM初始化

RM的初始化与TM基本类似，所不同的是，RMClient.init方法中创建的对象是RmNettyRemotingClient。

public static void init(String applicationId, String transactionServiceGroup) {
        RmNettyRemotingClient rmNettyRemotingClient = RmNettyRemotingClient.getInstance(applicationId, transactionServiceGroup);
        //设置默认资源管理器
        rmNettyRemotingClient.setResourceManager(DefaultResourceManager.get());
        //设置事务处理器
        rmNettyRemotingClient.setTransactionMessageHandler(DefaultRMHandler.get());
        rmNettyRemotingClient.init();
    }

因为RmNettyRemotingClient和TmNettyRemotingClient都继承相同的父类，创建对象都是调用父类的构造方法，所以对象创建流程都是一样的。
还有一点不同的是，RmNettyRemotingClient注册的处理器不同。

private void registerProcessor() {
        // 1.registry rm client handle branch commit processor
        RmBranchCommitProcessor rmBranchCommitProcessor = new RmBranchCommitProcessor(getTransactionMessageHandler(), this);
        super.registerProcessor(MessageType.TYPE_BRANCH_COMMIT, rmBranchCommitProcessor, messageExecutor);
        // 2.registry rm client handle branch commit processor
        RmBranchRollbackProcessor rmBranchRollbackProcessor = new RmBranchRollbackProcessor(getTransactionMessageHandler(), this);
        super.registerProcessor(MessageType.TYPE_BRANCH_ROLLBACK, rmBranchRollbackProcessor, messageExecutor);
        // 3.registry rm handler undo log processor
        RmUndoLogProcessor rmUndoLogProcessor = new RmUndoLogProcessor(getTransactionMessageHandler());
        super.registerProcessor(MessageType.TYPE_RM_DELETE_UNDOLOG, rmUndoLogProcessor, messageExecutor);
        // 4.registry TC response processor
        ClientOnResponseProcessor onResponseProcessor =
            new ClientOnResponseProcessor(mergeMsgMap, super.getFutures(), getTransactionMessageHandler());
        super.registerProcessor(MessageType.TYPE_SEATA_MERGE_RESULT, onResponseProcessor, null);
        super.registerProcessor(MessageType.TYPE_BRANCH_REGISTER_RESULT, onResponseProcessor, null);
        super.registerProcessor(MessageType.TYPE_BRANCH_STATUS_REPORT_RESULT, onResponseProcessor, null);
        super.registerProcessor(MessageType.TYPE_GLOBAL_LOCK_QUERY_RESULT, onResponseProcessor, null);
        super.registerProcessor(MessageType.TYPE_REG_RM_RESULT, onResponseProcessor, null);
        // 5.registry heartbeat message processor
        ClientHeartbeatProcessor clientHeartbeatProcessor = new ClientHeartbeatProcessor();
        super.registerProcessor(MessageType.TYPE_HEARTBEAT_MSG, clientHeartbeatProcessor, null);
    }

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/weixin_38308374/article/details/108779984

内容来源于网络，如有侵权，请联系作者删除！

Javamysql分布式事务seata

相关文章

    查看更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com