Seata解析

文章23 |   阅读 16341 |   点赞0

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

Seata解析-数据源代理DataSourceProxy详解

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

本文基于seata 1.3.0版本

前面通过十多篇文章详细介绍了TC端。从这篇文章开始介绍RM。
RM是资源管理器，资源指的就是数据库，RM主要与分支事务有关。RM会处理业务数据。
在《Seata解析-seata部署启动初体验》中，使用了类DataSourceProxy创建数据源代理。这里DataSourceProxy代理的就是业务数据库的数据源。因此本文从DataSourceProxy开始，开启分析RM的旅程。
下图是DataSourceProxy的继承结构：

DataSourceProxy实现了Resource接口，说明DataSourceProxy自身也是一种资源，可以被资源管理器ResourceManager管理。抽象类AbstractDataSourceProxy对DataSource提供了一部分很简单的方法实现，重要方法都是在DataSourceProxy中实现的。
下面具体看一下DataSourceProxy中方法实现。

一、构造方法实现

DataSourceProxy提供了两个构造方法：

public DataSourceProxy(DataSource targetDataSource) {
        this(targetDataSource, DEFAULT_RESOURCE_GROUP_ID);
    }
	//第二个参数在1.3.0版本里面没有使用，或者说设置这个值没有意义
	//所以这里使用默认值即可。
    public DataSourceProxy(DataSource targetDataSource, String resourceGroupId) {
        super(targetDataSource);
        init(targetDataSource, resourceGroupId);
    }
    private void init(DataSource dataSource, String resourceGroupId) {
        this.resourceGroupId = resourceGroupId;
        try (Connection connection = dataSource.getConnection()) {
            //jdbcUrl是我们自己配置的数据库连接
            jdbcUrl = connection.getMetaData().getURL();
            //从url中分析当前使用的是什么数据库，可能是oracle、mysql等
            //默认使用druid中的JdbcUtils.getDbType()分析
            //分析规则很简单，通过对url前缀匹配得到，比如mysql数据库连接前缀是“jdbc:mysql:”
            dbType = JdbcUtils.getDbType(jdbcUrl);
            if (JdbcConstants.ORACLE.equals(dbType)) {
                userName = connection.getMetaData().getUserName();
            }
        } catch (SQLException e) {
            throw new IllegalStateException("can not init dataSource", e);
        }
        //资源管理器：DefaultResourceManager
        //DataSourceProxy实现了Resource接口，因此本类就是一个资源
        //下面的代码向资源管理器管理注册本类
        DefaultResourceManager.get().registerResource(this);
        //ENABLE_TABLE_META_CHECKER_ENABLE的作用：是否开启定时任务，用于定时将表结构缓存在本地内存
        //默认1分钟运行一次。
        //缓存的表结构在RM保存记录快照时使用，如果内存中没有缓存，会实时查询数据库。
        if (ENABLE_TABLE_META_CHECKER_ENABLE) {
            tableMetaExcutor.scheduleAtFixedRate(() -> {
                try (Connection connection = dataSource.getConnection()) {
                    TableMetaCacheFactory.getTableMetaCache(DataSourceProxy.this.getDbType())
                        .refresh(connection, DataSourceProxy.this.getResourceId());
                } catch (Exception ignore) {
                }
            }, 0, TABLE_META_CHECKER_INTERVAL, TimeUnit.MILLISECONDS);
        }
    }

init()是DataSourceProxy的一个重要方法，其主要完成下面三个任务：

  1. 判断使用的是什么数据库，比如mysql，oracle；
  2. 向资源管理器注册DataSourceProxy，之所以可以注册，是因为DataSourceProxy实现了Resource接口，注册的主要工作是找到与事务分组对应的TC集群，并与集群中的每台机器建立连接；
  3. 判断是否启动定时任务，定时任务的作用是缓存数据库表结构，表结构在RM保存数据快照的时候使用。

第三个任务在介绍RM保存快照的时候在详细说明。下面重点看一下第二个任务，也就是下面这行代码所做的事情：

//将DataSourceProxy注册到默认资源管理器DefaultResourceManager中
DefaultResourceManager.get().registerResource(this);

上面代码调用了DefaultResourceManager的registerResource方法：

//入参是DataSourceProxy对象
	public void registerResource(Resource resource) {
        //resource.getBranchType()返回AT，
        //从这里可以看出DataSourceProxy只适用于AT模式
        //getResourceManager()返回的是DataSourceManager，
        //DataSourceManager也是用于AT模式下
        getResourceManager(resource.getBranchType()).registerResource(resource);
    }

DefaultResourceManager相当于一个路由类，它有一个Map属性resourceManagers，里面保存了每种模式对应的资源管理器，我们使用的是AT模式，因此getResourceManager方法从resourceManagers中取出AT模式的资源管理器，也就是DataSourceManager对象，然后调用DataSourceManager的registerResource方法。

public void registerResource(Resource resource) {
        DataSourceProxy dataSourceProxy = (DataSourceProxy) resource;
        //dataSourceProxy.getResourceId()返回的是我们在程序中设置的数据库连接，
        //不过如果连接中有“？”，它会把问号后面的内容去掉
        //比如：在应用程序中设置数据库连接为jdbc:mysql://localhost:3306/test?characterEncoding=utf8
        //dataSourceProxy.getResourceId()实际返回jdbc:mysql://localhost:3306/test
        dataSourceCache.put(dataSourceProxy.getResourceId(), dataSourceProxy);
        super.registerResource(dataSourceProxy);
    }

上面代码的最后调用父类的registerResource方法：

public void registerResource(Resource resource) {
        //向TC注册资源
        //下面代码首先获得RmNettyRemotingClient实例，
        //这里获取时，实例其实已经创建完毕，创建是在另一个初始化过程中完成的
        //后面的文章详细介绍这个过程
        //现在只需要知道RmNettyRemotingClient已经启动客户端Netty，
        //并且创建了连接池，不过连接池中还没有连接
        RmNettyRemotingClient.getInstance().registerResource(resource.getResourceGroupId(), resource.getResourceId());
    }

上面方法最后又去调用实例RmNettyRemotingClient的registerResource方法：

public void registerResource(String resourceGroupId, String resourceId) {
        //启动的时候，因为RM还没有建立与服务端的连接，所以下面的if判断是true
        //getClientChannelManager()返回NettyClientChannelManager对象，
        //NettyClientChannelManager就是上面方法提到的连接池，它管理与TC的连接，
        //该连接池是在RmNettyRemotingClient的构造方法中创建的，但是创建时不会建立与TC的连接
        if (getClientChannelManager().getChannels().isEmpty()) {
        	//下面reconnect方法的入参是分组事务名，也就是配置文件中spring.cloud.alibaba.seata.tx-service-group的值
        	//reconnect方法用于创建与TC的连接
            getClientChannelManager().reconnect(transactionServiceGroup);
            return;
        }
        synchronized (getClientChannelManager().getChannels()) {
            for (Map.Entry<String, Channel> entry : getClientChannelManager().getChannels().entrySet()) {
                String serverAddress = entry.getKey();
                Channel rmChannel = entry.getValue();
                if (LOGGER.isInfoEnabled()) {
                    LOGGER.info("will register resourceId:{}", resourceId);
                }
                sendRegisterMessage(serverAddress, rmChannel, resourceId);
            }
        }
    }

NettyClientChannelManager的reconnect方法根据事务分组从注册中心找到提供服务的TC集群，并获得集群中每台机器的地址，接着创建与每台机器的连接。下面具体看一下代码：

void reconnect(String transactionServiceGroup) {
        List<String> availList = null;
        try {
            //获得与事务分组对应的集群中每台机器地址
            availList = getAvailServerList(transactionServiceGroup);
        } catch (Exception e) {
            LOGGER.error("Failed to get available servers: {}", e.getMessage(), e);
            return;
        }
        //如果集群中没有机器提供服务，那么打印出日志，seata有一个定时任务，
        //每过一段时间会重新查看集群中是否有机器
        if (CollectionUtils.isEmpty(availList)) {
            String serviceGroup = RegistryFactory.getInstance()
                                                 .getServiceGroup(transactionServiceGroup);
            LOGGER.error("no available service '{}' found, please make sure registry config correct", serviceGroup);
            return;
        }
        //遍历每台机器
        for (String serverAddress : availList) {
            try {
                //建立与TC的连接
                acquireChannel(serverAddress);
            } catch (Exception e) {
                LOGGER.error("{} can not connect to {} cause:{}",FrameworkErrorCode.NetConnect.getErrCode(), serverAddress, e.getMessage(), e);
            }
        }
    }
    private List<String> getAvailServerList(String transactionServiceGroup) throws Exception {
        //根据服务分组名首先获得集群名，然后根据集群名查询得到提供服务的TC端机器列表
        //seata允许TC端多机部署，将TC端的多台机器分为一个集群，并给集群一个名字，一个服务分组对应一个集群
        List<InetSocketAddress> availInetSocketAddressList = RegistryFactory.getInstance()
                                                                            .lookup(transactionServiceGroup);
        if (CollectionUtils.isEmpty(availInetSocketAddressList)) {
            return Collections.emptyList();
        }
        //将availInetSocketAddressList中地址转化为IP:PORT的形式
        return availInetSocketAddressList.stream()
                                         .map(NetUtil::toStringAddress)
                                         .collect(Collectors.toList());
    }
    Channel acquireChannel(String serverAddress) {
        //channels是一个Map<String, Channel>对象，通过该属性可以看出，每个TC端，RM只维持一个连接
        //在RM启动的时候，channels里面没有任何连接，所以channelToServer=null
        Channel channelToServer = channels.get(serverAddress);
        if (channelToServer != null) {
            //getExistAliveChannel()检查当前连接是否可用，如果不可用返回null
            channelToServer = getExistAliveChannel(channelToServer, serverAddress);
            if (channelToServer != null) {
                return channelToServer;
            }
        }
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("will connect to " + serverAddress);
        }
        channelLocks.putIfAbsent(serverAddress, new Object());
        //这个位置做了并发控制，每次只能一个线程创建与TC的连接
        synchronized (channelLocks.get(serverAddress)) {
            return doConnect(serverAddress);
        }
    }
    private Channel doConnect(String serverAddress) {
        Channel channelToServer = channels.get(serverAddress);
        if (channelToServer != null && channelToServer.isActive()) {
            return channelToServer;
        }
        Channel channelFromPool;
        try {
            //poolKeyFunction.apply调用的是RmNettyRemotingClient.getPoolKeyFunction()方法
            //该方法创建RegisterRMRequest对象和NettyPoolKey对象，
            //RegisterRMRequest对象在建立与TC连接后，会把该对象发送到TC进行注册
            NettyPoolKey currentPoolKey = poolKeyFunction.apply(serverAddress);
            NettyPoolKey previousPoolKey = poolKeyMap.putIfAbsent(serverAddress, currentPoolKey);
            if (previousPoolKey != null && previousPoolKey.getMessage() instanceof RegisterRMRequest) {
                RegisterRMRequest registerRMRequest = (RegisterRMRequest) currentPoolKey.getMessage();
                ((RegisterRMRequest) previousPoolKey.getMessage()).setResourceIds(registerRMRequest.getResourceIds());
            }
            //调用NettyPoolableFactory的makeObject方法创建与TC的连接
            channelFromPool = nettyClientKeyPool.borrowObject(poolKeyMap.get(serverAddress));
            channels.put(serverAddress, channelFromPool);
        } catch (Exception exx) {
            LOGGER.error("{} register RM failed.",FrameworkErrorCode.RegisterRM.getErrCode(), exx);
            throw new FrameworkException("can not register RM,err:" + exx.getMessage());
        }
        return channelFromPool;
    }

上面的代码根据事务分组查找机器的逻辑是：首先从file.conf文件查找“service.vgroupMapping.事务分组”的配置，该配置就是TC集群的名字，如果注册中心使用的是zk，该配置也是zk上的路径，所以接下来，seata访问zk，查找路径：/registry/zk/TC集群名下的value，这个value就是机器列表。得到TC的机器列表后，下面就要与TC建立连接，建立连接其实是委托给连接池去完成了，这里不介绍如何委托过去的，有兴趣的可以看一下Apache的GenericKeyedObjectPool。连接池创建连接时最终是调用的NettyPoolableFactory的makeObject方法：

public Channel makeObject(NettyPoolKey key) {
        InetSocketAddress address = NetUtil.toInetSocketAddress(key.getAddress());
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("NettyPool create channel to " + key);
        }
        //通过Netty的客户端获得与TC的连接
        Channel tmpChannel = clientBootstrap.getNewChannel(address);
        long start = System.currentTimeMillis();
        Object response;
        Channel channelToServer = null;
        if (key.getMessage() == null) {
            throw new FrameworkException("register msg is null, role:" + key.getTransactionRole().name());
        }
        try {
            //向TC发送RegisterRMRequest请求
            //也就是向TC注册RM
            response = rpcRemotingClient.sendSyncRequest(tmpChannel, key.getMessage());
            if (!isRegisterSuccess(response, key.getTransactionRole())) {
                //如果TC返回失败，下面的方法会构建失败信息，并抛出异常
                rpcRemotingClient.onRegisterMsgFail(key.getAddress(), tmpChannel, response, key.getMessage());
            } else {
                //如果成功，则将与TC的连接注册到NettyClientChannelManager的channels属性中
                channelToServer = tmpChannel;
                rpcRemotingClient.onRegisterMsgSuccess(key.getAddress(), tmpChannel, response, key.getMessage());
            }
        } catch (Exception exx) {
            if (tmpChannel != null) {
                tmpChannel.close();
            }
            throw new FrameworkException(
                "register " + key.getTransactionRole().name() + " error, errMsg:" + exx.getMessage());
        }
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("register success, cost " + (System.currentTimeMillis() - start) + " ms, version:" + getVersion(
                response, key.getTransactionRole()) + ",role:" + key.getTransactionRole().name() + ",channel:"
                + channelToServer);
        }
        return channelToServer;
    }

到这里DataSourceProxy的注册流程全部结束，可以看到注册最终是建立与TC的连接，然后发送RegisterRMRequest请求注册RM。
注册的流程相对还是比较复杂的。
最后在说一点是如何定义TC属于哪个集群？
在TC的register.conf文件中，在配置registry下的属性时，可以看到部分注册中心有cluster的设置，这个cluster就表示了当前TC实例属于哪个集群。TC启动后也会在注册中心对应的目录下添加自己本机的IP地址。这样RM就可以从注册中心找到TC了。

二、getConnection()方法

DataSourceProxy中除了init方法之外，我们最后再看一下getConnection()方法：

public ConnectionProxy getConnection() throws SQLException {
        Connection targetConnection = targetDataSource.getConnection();
        return new ConnectionProxy(this, targetConnection);
    }
    @Override
    public ConnectionProxy getConnection(String username, String password) throws SQLException {
        Connection targetConnection = targetDataSource.getConnection(username, password);
        return new ConnectionProxy(this, targetConnection);
    }

DataSourceProxy对getConnection重载了，getConnection的作用是返回数据库连接，从上面代码可以看出seata使用ConnectionProxy对真实的数据库连接进行了代理，所以上层应用使用的都是连接的代理对象。关于代理对象后面的文章会详细介绍。

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

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

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

Javasql分布式事务seata

相关文章

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