目录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获取数据表元数据的资料请关注我们其它相关文章！
加载中,请稍侯......
精彩评论