前言
我们都知道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()) .refresh(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 SQLException { 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); 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遵循了读多写少用缓存的原则，来尽可能地降低该操作带来的影响。
作者：梦想实现家_Z
链接：https://juejin.cn/post/7164649327913074702
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
