開篇
?這篇文章的目的是講解RM Executor模塊當中一些通用的方法，這些方法在各個Executor的父類當中實現(xiàn)的，各個子類Executor模塊都會復(fù)用，因此抽取出來統(tǒng)一的進行講解。
?個人是認為抽取通用的內(nèi)容放在一篇文章講解完后可以針對每類Executor講解特有的功能，這樣能夠有更好的理解。這篇文章講解Executor的父類BaseTransactionalExecutor。
類依賴圖
說明：
- 著重講解BaseTransactionalExecutor抽象父類。
BaseTransactionalExecutor通用方法介紹
public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor {
protected StatementProxy<S> statementProxy;
protected StatementCallback<T, S> statementCallback;
protected SQLRecognizer sqlRecognizer;
private TableMeta tableMeta;
public BaseTransactionalExecutor(StatementProxy<S> statementProxy,
StatementCallback<T, S> statementCallback, SQLRecognizer sqlRecognizer) {
this.statementProxy = statementProxy;
this.statementCallback = statementCallback;
this.sqlRecognizer = sqlRecognizer;
}
}
說明：
- BaseTransactionalExecutor的核心變量SQLRecognizer sqlRecognizer。
- SQLRecognizer sqlRecognizer的是通過構(gòu)造函數(shù)進行賦值的。
- SQLRecognizer是由SQLVisitorFactory的工廠方法返回生成的。
public class SQLVisitorFactory {
public static SQLRecognizer get(String sql, String dbType) {
List<SQLStatement> asts = SQLUtils.parseStatements(sql, dbType);
if (asts == null || asts.size() != 1) {
throw new UnsupportedOperationException("Unsupported SQL: " + sql);
}
SQLRecognizer recognizer = null;
SQLStatement ast = asts.get(0);
if (JdbcConstants.MYSQL.equalsIgnoreCase(dbType)) {
if (ast instanceof SQLInsertStatement) {
recognizer = new MySQLInsertRecognizer(sql, ast);
} else if (ast instanceof SQLUpdateStatement) {
recognizer = new MySQLUpdateRecognizer(sql, ast);
} else if (ast instanceof SQLDeleteStatement) {
recognizer = new MySQLDeleteRecognizer(sql, ast);
} else if (ast instanceof SQLSelectStatement) {
if (((SQLSelectStatement)ast).getSelect().getQueryBlock().isForUpdate()) {
recognizer = new MySQLSelectForUpdateRecognizer(sql, ast);
}
}
} else {
throw new UnsupportedOperationException("Just support MySQL by now!");
}
return recognizer;
}
}
說明：
- SQLRecognizer 根據(jù)不同的SQL類型生成的。
- MySQLInsertRecognizer MySQLUpdateRecognizer MySQLDeleteRecognizer MySQLSelectForUpdateRecognizer。
- SQLUtils.parseStatements作為druid開源包提供的功能負責返回SQLStatement對象。
- SQLStatement是根據(jù)不同SQL生產(chǎn)的SQL會話對象包含SQL含有的一些通用信息。
public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor {
protected String buildWhereConditionByPKs(List<Field> pkRows) throws SQLException {
StringBuffer whereConditionAppender = new StringBuffer();
for (int i = 0; i < pkRows.size(); i++) {
Field field = pkRows.get(i);
whereConditionAppender.append(getColumnNameInSQL(field.getName()) + " = ?");
if (i < (pkRows.size() - 1)) {
whereConditionAppender.append(" OR ");
}
}
return whereConditionAppender.toString();
}
}
說明：
- 根據(jù)primary key生成where條件語句。
- 根據(jù)主鍵的個數(shù)構(gòu)建 select x from table where c1=? OR c2=?格式。
public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor {
protected String getColumnNameInSQL(String columnName) {
String tableAlias = sqlRecognizer.getTableAlias();
if (tableAlias == null) {
return columnName;
} else {
return tableAlias + "." + columnName;
}
}
}
說明：
- 獲取SQL當中的列名，通過sqlRecognizer獲取別名。
- 如果有別名那么列名就是 tableAlias.columnName，否則就是columnName。
public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor {
protected String getFromTableInSQL() {
String tableName = sqlRecognizer.getTableName();
String tableAlias = sqlRecognizer.getTableAlias();
if (tableAlias == null) {
return tableName;
} else {
return tableName + " " + tableAlias;
}
}
}
說明：
- 獲取SQL當中的表名，通過sqlRecognizer獲取別名。
- 如果有別名那么列名就是 tableName tableAlias，否則就是tableName。
- 類似select * from tableName tableAlias。
public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor {
protected TableMeta getTableMeta() {
return getTableMeta(sqlRecognizer.getTableName());
}
protected TableMeta getTableMeta(String tableName) {
if (tableMeta != null) {
return tableMeta;
}
tableMeta = TableMetaCache.getTableMeta(statementProxy.getConnectionProxy().getDataSourceProxy(), tableName);
return tableMeta;
}
}
說明：
- 獲取Table的Meta數(shù)據(jù)，通過TableMetaCache.getTableMeta()操作實現(xiàn)。
- TableMetaCache.getTableMeta()是實現(xiàn)Cache功能的元數(shù)據(jù)獲取功能。
public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor {
protected String buildLockKey(TableRecords rowsIncludingPK) {
if (rowsIncludingPK.size() == 0) {
return null;
}
StringBuilder sb = new StringBuilder();
sb.append(rowsIncludingPK.getTableMeta().getTableName());
sb.append(":");
boolean flag = false;
for (Field field : rowsIncludingPK.pkRows()) {
if (flag) {
sb.append(",");
} else {
flag = true;
}
sb.append(field.getValue());
}
return sb.toString();
}
}
說明：
- buildLockKey實現(xiàn)鎖定key的生成邏輯，表名+主鍵列名的拼接。
- buildLockKey的實現(xiàn)邏輯：tableName:pkName1,pkName2.
public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor {
protected void prepareUndoLog(TableRecords beforeImage, TableRecords afterImage) throws SQLException {
if (beforeImage.getRows().size() == 0 && afterImage.getRows().size() == 0) {
return;
}
ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
TableRecords lockKeyRecords = sqlRecognizer.getSQLType() == SQLType.DELETE ? beforeImage : afterImage;
String lockKeys = buildLockKey(lockKeyRecords);
connectionProxy.appendLockKey(lockKeys);
SQLUndoLog sqlUndoLog = buildUndoItem(beforeImage, afterImage);
connectionProxy.appendUndoLog(sqlUndoLog);
}
protected SQLUndoLog buildUndoItem(TableRecords beforeImage, TableRecords afterImage) {
SQLType sqlType = sqlRecognizer.getSQLType();
String tableName = sqlRecognizer.getTableName();
SQLUndoLog sqlUndoLog = new SQLUndoLog();
sqlUndoLog.setSqlType(sqlType);
sqlUndoLog.setTableName(tableName);
sqlUndoLog.setBeforeImage(beforeImage);
sqlUndoLog.setAfterImage(afterImage);
return sqlUndoLog;
}
}
說明：
- prepareUndoLog負責生產(chǎn)待回滾記錄的日志，按照執(zhí)行前鏡像和執(zhí)行后鏡像對比生成。
- 回滾日志的存儲對象是類SQLUndoLog。
public class TableMetaCache {
public static TableMeta getTableMeta(DataSourceProxy dataSourceProxy, String tableName) {
return getTableMeta(dataSourceProxy.getTargetDataSource(), tableName);
}
public static TableMeta getTableMeta(final DruidDataSource druidDataSource, final String tableName) {
String dataSourceKey = druidDataSource.getUrl();
TableMeta tmeta = null;
final String key = dataSourceKey + "." + tableName;
try {
tmeta = TABLE_META_CACHE.get(key, new Callable<TableMeta>() {
@Override
public TableMeta call() throws Exception {
return fetchSchema(druidDataSource, tableName);
}
});
} catch (ExecutionException e) {
}
if (tmeta == null) {
try {
tmeta = fetchSchema(druidDataSource, tableName);
} catch (SQLException e) {
}
}
return tmeta;
}
private static TableMeta fetchSchema(DruidDataSource druidDataSource, String tableName) throws SQLException {
return fetchSchemeInDefaultWay(druidDataSource, tableName);
}
private static TableMeta fetchSchemeInDefaultWay(DruidDataSource druidDataSource, String tableName)
throws SQLException {
Connection conn = null;
java.sql.Statement stmt = null;
java.sql.ResultSet rs = null;
try {
conn = druidDataSource.getConnection();
stmt = conn.createStatement();
StringBuffer sb = new StringBuffer("SELECT * FROM " + tableName + " LIMIT 1");
rs = stmt.executeQuery(sb.toString());
ResultSetMetaData rsmd = rs.getMetaData();
DatabaseMetaData dbmd = conn.getMetaData();
return resultSetMetaToSchema(rsmd, dbmd, tableName);
} catch (Exception e) {
if (e instanceof SQLException) {
throw ((SQLException)e);
}
throw new SQLException("Failed to fetch schema of " + tableName, e);
} finally {
}
}
private static TableMeta resultSetMetaToSchema(ResultSetMetaData rsmd, DatabaseMetaData dbmd, String tableName)
throws SQLException {
String schemaName = rsmd.getSchemaName(1);
String catalogName = rsmd.getCatalogName(1);
TableMeta tm = new TableMeta();
tm.setTableName(tableName);
java.sql.ResultSet rs1 = dbmd.getColumns(catalogName, schemaName, tableName, "%");
while (rs1.next()) {
ColumnMeta col = new ColumnMeta();
col.setTableCat(rs1.getString("TABLE_CAT"));
col.setTableSchemaName(rs1.getString("TABLE_SCHEM"));
col.setTableName(rs1.getString("TABLE_NAME"));
col.setColumnName(rs1.getString("COLUMN_NAME"));
col.setDataType(rs1.getInt("DATA_TYPE"));
col.setDataTypeName(rs1.getString("TYPE_NAME"));
col.setColumnSize(rs1.getInt("COLUMN_SIZE"));
col.setDecimalDigits(rs1.getInt("DECIMAL_DIGITS"));
col.setNumPrecRadix(rs1.getInt("NUM_PREC_RADIX"));
col.setNullAble(rs1.getInt("NULLABLE"));
col.setRemarks(rs1.getString("REMARKS"));
col.setColumnDef(rs1.getString("COLUMN_DEF"));
col.setSqlDataType(rs1.getInt("SQL_DATA_TYPE"));
col.setSqlDatetimeSub(rs1.getInt("SQL_DATETIME_SUB"));
col.setCharOctetLength(rs1.getInt("CHAR_OCTET_LENGTH"));
col.setOrdinalPosition(rs1.getInt("ORDINAL_POSITION"));
col.setIsNullAble(rs1.getString("IS_NULLABLE"));
col.setIsAutoincrement(rs1.getString("IS_AUTOINCREMENT"));
tm.getAllColumns().put(col.getColumnName(), col);
}
java.sql.ResultSet rs2 = dbmd.getIndexInfo(catalogName, schemaName, tableName, false, true);
String indexName = "";
while (rs2.next()) {
indexName = rs2.getString("INDEX_NAME");
String colName = rs2.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(rs2.getBoolean("NON_UNIQUE"));
index.setIndexQualifier(rs2.getString("INDEX_QUALIFIER"));
index.setIndexName(rs2.getString("INDEX_NAME"));
index.setType(rs2.getShort("TYPE"));
index.setOrdinalPosition(rs2.getShort("ORDINAL_POSITION"));
index.setAscOrDesc(rs2.getString("ASC_OR_DESC"));
index.setCardinality(rs2.getInt("CARDINALITY"));
index.getValues().add(col);
if ("PRIMARY".equalsIgnoreCase(indexName) || indexName.equalsIgnoreCase(
rsmd.getTableName(1) + "_pkey")) {
index.setIndextype(IndexType.PRIMARY);
} else if (index.isNonUnique() == false) {
index.setIndextype(IndexType.Unique);
} else {
index.setIndextype(IndexType.Normal);
}
tm.getAllIndexes().put(indexName, index);
}
}
IndexMeta index = tm.getAllIndexes().get(indexName);
if (index.getIndextype().value() != 0) {
if ("H2 JDBC Driver".equals(dbmd.getDriverName())) {
if (indexName.length() > 11 && "PRIMARY_KEY".equalsIgnoreCase(indexName.substring(0, 11))) {
index.setIndextype(IndexType.PRIMARY);
}
} else if (dbmd.getDriverName() != null && dbmd.getDriverName().toLowerCase().indexOf("postgresql") >= 0) {
if ((tableName + "_pkey").equalsIgnoreCase(indexName)) {
index.setIndextype(IndexType.PRIMARY);
}
}
}
return tm;
}
}
說明：
- TableMetaCache提供獲取表數(shù)據(jù)的功能，包括緩存功能。
- 元數(shù)據(jù)功能包括數(shù)據(jù)庫本身元數(shù)據(jù)和返回結(jié)果元數(shù)據(jù)。
- 通過DataSource->Connection->Statement的邏輯執(zhí)行SQL語句獲取ResultSetMetaData結(jié)果元數(shù)據(jù)。
- 通過Connection獲取DatabaseMetaData dbmd數(shù)據(jù)庫元數(shù)據(jù)
