Seata AT获取数据表元数据源码详解

目录javascript

• 前言
• 一探究竟
• 小结

前言

我们都知道Seata AT是基于前后镜像来实现事务的成功回滚的，前后镜像的生成依赖于数据表的元数据，Seata是如何生成前后镜像的可以看这篇博客:你知道Seata AT模式中前后镜像是如何生成的嘛？。

起初我以为数据库Driver提供了现成的API给开发人员获取指定数据表的元数据，今天看了源码才知道，并没有想象中那么简单。下面我们就来一起看看到底是怎么一回事儿。

一探究竟

我们直接展开关键性的seata源码，进入DataSourceProxy.init()方法中：

// 是否允许开启定时任务检查更新元数据
if (ENABLE_TABLE_META_CHECKER_ENABLE) {
    // 开启定时任务，默认一分钟更新检查一下
    tableMetaExecutor.scheduleAtFixedRate(() -> {
        // 获取数据库链接
        try (Connection connection = dataSource.getConnection()) {
             // 更新缓存中的数据表元数据
             TableMetaCacheFactory.getTableMetaCache(DataSourceProxy.this.getDbType())
                        .rwww.开发者_开发入门devze.comefresh(connection, DataSourceProxy.this.getResourceId());
        } catch (Exception ignore) {}
   }, 0, TABLE_META_CHECKER_INTERVAL, TimeUnit.MILLISECONDS);
}

Seata AT在创建了DataSourceProxy对象后，马上会启动一个定时任务，一分钟检查一次缓存中的元数据。

跟着关键代码，我们可以追踪到AbstractTableMetaCache类，这个抽象类其实就提供了两个方法：

public abstract class AbstractTableMetaCache implements TableMetaCache {
    @Override
    public TableMeta getTableMeta(final Connection connection, final String tableName, String resourceId) {
        // 如果缓存中有对应数据就返回，否则就去查询元数据并放在缓存中。
    }
    @Override
    public void refresh(final Connection connection, String resourceId) {
        // 更新缓存
    }
}

最后我们发现获取数据表元数据的代码实现在fetchSchema()方法中，但是这个方法是一个抽象方法，有多个实现：

我们就挑一个mysqlTableMetaCache来看一下里面是如何实现的。

@Override
    protected TableMeta fetchSchema(Connection connection, String tableName) throws SQLExceptionphp {
        String sql = "SELECT * FROM " + ColumnUtils.addEscape(tableName, JdbcConstants.MYSQL) + " LIMIT 1";
        try (Statement stmt = connection.createStatement();
            // 执行SQL语句：SELECT * FROM [tableName] LIMIT 1；
            ResultSet rs = stmt.executeQuery(sql)) {
            // 根据执行结果获取元数据
            return resultSetMetaToSchema(rs.getMetaData(), connection.getMetaData());
        } catch (SQLException sqlEx) {
            throw sqlEx;
        } catch (Exception e) {
            throw new SQLException(String.format("Failed to fetch schema of %s", tableName), e);
        }
    }

根据上面源码，我们发现Seata获取Mysql数据表的元数据竟然是通过SELECT * FROM [tableName] LIMIT 1来的，但是事实并不是我们想象的这么简单，继续深入resultSetMetaToSchema()方法：

private TableMeta resultSetMetaToSchema(ResultSetMetaData rsmd, DatabaseMetaData dbmd)
        throws SQLException {
        //always "" for mysql
        String schemaName = rsmd.getSchemaName(1);
        String catalogName = rsmd.getCatalogName(1);
        /*
         * 通过ResultSetMetaData获取tableName可以避免以下情况
         *
         * select * from account_tbl
         * select * from account_TBL
         * select * from `account_tbl`
         * select * from account.account_tbl
         */
        String tableName = rsmd.getTableName(1);
        TableMeta tm = new TableMeta();
        tm.setTableName(tableName);
        /*
         * here has two different type to get the data
         * make sure the table name was right
         * 1. show full columns from xxx from xxx(normal)
         * 2. select xxx from xxx where catalog_name like ? and table_name like ?(informationSchema=true)
         */
        // 通过dbmd发送查询语句获取指定表中的所有列信息
        try (ResultSet rsColumns = dbmd.getColumns(catalogName, schemaName, tableName, "%");
             // 发送查询语句获取表中索引信息
             ResultSet rsIndex = dbmd.getIndexInfo(catalogName, schemaName, tableName, false, true);
  编程客栈           // 查询更新行中的任何值时自动更新的列的信息
             ResultSet onUpdateColumns = dbmd.getVersionColumns(catalogName, schemaName, tableName)) {
            // 收集列信息
            while (rsColumns.next()) {
                ColumnMeta col = new ColumnMeta();
                col.setTableCat(rsColumns.getString("TABLE_CAT"));
                col.setTableSchemaName(rsColumns.getString("TABLE_SCHEM"));
                col.setTableName(rsColumns.getString("TABLE_NAME"));
                col.setColumnName(rsColumns.getString("COLUMN_NAME"));
                col.setDataType(rsColumns.getInt("DATA_TYPE"));
                col.setDataTypeName(rsColumns.getString("TYPE_NAME"));
                col.setColumnSize(rsColumns.getInt("COLUMN_SIZE"));
                col.setDecimalDigits(rsColumns.getInt("DECIMAL_DIGITS"));
                col.setNumPrecRadix(rsColumns.getInt("NUM_PREC_RADIX"));
                col.setNullAble(rsColumns.getInt("NULLABLE"));
                col.setRemarks(rsColumns.getString("REMARKS"));
                col.setColumnDef(rsColumns.getString("COLUMN_DEF"));
                col.setSqlDataType(rsColumns.getInt("SQL_DATA_TYPE"));
                col.setSqlDatetimeSub(rsColumns.getInt("SQL_DATETIME_SUB"));
                col.setCharOctetLength(rsColumns.getInt("CHAR_OCTET_LENGTH"));
                col.setOrdinalPosition(rsColumns.getInt("ORDINAL_POSITION"));
                col.setIsNullAble(rsColumns.getString("IS_NULLABLE"));
                col.setIsAutoincrement(rsColumns.getString("IS_AUTOINCREMENT"));
                if (tm.getAllColumns().containsKey(col.getColumnName())) {
                    throw new NotSupportYetException("Not support the table has the same column name with different case yet");
                }
                tm.getAllColumns().put(col.getColumnName(), col);
            }
            while (onUpdateColumns.next()) {
                tm.getAllColumns().get(onUpdateColumns.getString("COLUMN_NAME")).setOnUpdate(true);
            }
            // 收集索引信息
            while (rsIndex.next()) {
                String indexName = rsIndex.getString("INDEX_NAME");
                String colName = rsIndex.getString("COLUMN_NAME");
                ColumnMeta col = tm.getAllColumns().get(colName);
                if (tm.getAllIndexes().containsKey(indexName)) {
                    IndexMeta index = tm.getAllIndexes().get(indexName);
       android             index.getValues().add(col);
                } else {
                    IndexMeta index = new IndexMeta();
                    index.setIndexName(indexName);
                    index.setNonUnique(rsIndex.getBoolean("NON_UNIQUE"));
                    index.setIndexQualifier(rsIndex.getString("INDEX_QUALIFIER"));
                    index.setIndexName(rsIndex.getString("INDEX_NAME"));
                    index.setType(rsIndex.getShort("TYPE"));
                    index.setOrdinalPosition(rsIndex.getShort("ORDINAL_POSITION"));
                    index.setAscOrDesc(rsIndex.getString("ASC_OR_DESC"));
                    index.setCardinality(rsIndex.getInt("CARDINALITY"));
                    index.getValues().add(col);
                    if ("PRIMARY".equalsIgnoreCase(indexName)) {
                        index.setIndextype(IndexType.PRIMARY);
                    } else if (!index.isNonUnique()) {
                        index.setIndextype(IndexType.UNIQUE);
                    } else {
                        index.setIndextype(IndexType.NORMAL);
                    }
                    tm.getAllIndexes().put(indexName, index);
                }
            }
            if (tm.getAllIndexes().isEmpty()) {
                throw new ShouldNeverHappenException("Could not found any index in the table: " + tableName);
            }
        }
        return tm;
    }

可以发现，在mysql的实现中，我们查询一个表的元数据，需要执行四条SQL语句，另外oracle和PostgreSQL实现中，也是要执行三条查询语句的。在数据表变动不是很频繁的情况下，seata遵循读多写少用缓存的原则，并通过定时任务的方式来保持拿到的数据表元数据是最新的。

小结

在seata获取数据表元数据的实现中，我们通过阅读源码的方式，大致收获了以下几点：

1.seata AT模式默认会开启定时任务每分钟更新数据表元数据，这是一个配置项，在确认运行时数据表不会变更的情况下，开发人员可以不开启该定时任务关闭。client.rm.tableMetaCheckEnable=false即可关闭该定时任务。

2.seata获取数据表元数据至少需要进行三次以上的查询，这属于一个比较重的操作。为了避免获取元数据影响业务的吞吐量，seata遵循了读多写少用缓存的原则，来尽可能地降低该操作带来的影响。

以上就是Seata AT获取数据表元数据源码详解的详细内容，更多关于Seata AT获取数据表元数据的资料请关注我们其它相关文章！