首页 > 其他分享 >异构数据源同步之表结构同步 → 通过 jdbc 实现,没那么简单

异构数据源同步之表结构同步 → 通过 jdbc 实现,没那么简单

时间:2024-05-06 09:11:50浏览次数:23  
标签:COMMENT case 同步 return columnMeta 数据源 之表 类型 Types

开心一刻

今天坐沙发上看电视,旁边的老婆拿着手机贴了过来
老婆:老公,这次出门旅游,机票我准备买了哈
我:嗯
老婆:你、我、你爸妈、我爸妈,一共六张票
老婆:这上面还有意外保险,要不要买?
我:都特么团灭了,还买啥保险?

开心一刻

异构数据源同步

概念介绍

  • 数据源,不只是包含关系型数据库,还包括 NoSQL、数仓、中间件、ftp 等等,凡是有存储功能的都算
  • 异构,两端的数据源的结构存在差异,比如列数不一致、列类型不一致等等
  • 同步,将源数据源的数据同步到目标数据源,包括数据读取、转换和写入过程

所以,异构数据源同步就是指在不同类型或格式的数据源之间传输和同步数据的过程

同步策略

主要有两种同步策略:离线同步实时同步 ,各有其特点和适用场景
但是,这些我今天都不讲,就吊吊你们胃口

如果你们想了解,自己去查吧
今天我就要逆袭一把,将 离线同步 中的一个小配角转正成主角!

表结构同步

异构数据源同步 整个主线剧情中,数据同步 才是真正的主角

表结构同步 只能算活不过三集的那种配角

但今天不拍主线剧情,我要拍个番外篇来重点讲 表结构同步 ,我是导演嘛,当然我说了算

背景说明

主要是针对关系型数据库,当目标数据源的表不存在时,则先在目标数据源创建目标表,然后进行数据的同步

比如:从 MySQL 的表 tbl_t1 同步到 SQL Server 的表 tbl_tt ,若 tbl_tt 不存在,则根据 tbl_t1 的表结构创建 tbl_tt

所以这里就涉及到表结构的同步,也正是本文的主角!

如何实现

通过 jdbc 来实现,具体实现步骤如下

  1. 通过 jdbc 获取元数据信息:表元数据、列元数据、主键元数据、索引元数据

  2. 根据元数据拼接目标表的建表 SQL

  3. 通过 jdbc ,根据建表 SQL,在目标数据源创建目标表

第 3 步实现比较容易,难得是第 1、2步
虽然前路坑很多,但你们不要慌,我已经替你们趟掉很多了

我们以 MySQL 为例,假设我们库 test 下有表 tbl_sync

CREATE TABLE `tbl_sync` (
  `c_bigint_auto` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT 'bigint 类型',
  `c_bigint` bigint DEFAULT NULL COMMENT 'bigint 类型',
  `c_vachar` varchar(100) NOT NULL COMMENT 'varchar 类型',
  `c_char` char(32) NOT NULL COMMENT 'char 类型',
  `c_text` text NOT NULL COMMENT 'text 类型',
  `c_decimal_4` decimal(15,4) NOT NULL DEFAULT '5000.0000' COMMENT 'decimal 类型',
  `c_decimal_0` decimal(10,0) DEFAULT NULL COMMENT 'decimal 类型',
  `c_blob` blob COMMENT 'blob 类型',
  `c_bit` bit(1) DEFAULT NULL COMMENT 'bit 类型',
  `c_tinyint` tinyint DEFAULT NULL COMMENT 'tinyint 类型',
  `c_binary` binary(10) DEFAULT NULL COMMENT 'binary 类型',
  `c_float` float(13,0) DEFAULT NULL COMMENT 'float 类型',
  `c_double` double(23,0) DEFAULT NULL COMMENT 'double 类型',
  `c_varbinary` varbinary(20) DEFAULT NULL COMMENT 'varbinary 类型',
  `c_longblob` longblob COMMENT 'longblob 类型',
  `c_longtext` longtext COMMENT 'longtext 类型',
  `c_json` json DEFAULT NULL COMMENT 'json 类型',
  `c_date` date DEFAULT NULL COMMENT 'date 类型',
  `c_time` time(2) DEFAULT NULL COMMENT 'time 类型',
  `c_datetime` datetime(3) DEFAULT NULL COMMENT 'datetime 类型',
  `c_timestamp` timestamp(4) NULL DEFAULT NULL COMMENT 'timestamp 类型',
  `c_year` year DEFAULT NULL COMMENT 'year 类型',
  PRIMARY KEY (`c_vachar`,`c_char`,`c_bigint_auto`),
  UNIQUE KEY `uk_id` (`c_bigint_auto`),
  KEY `idx_name_salary` (`c_vachar`,`c_decimal_4`)
) COMMENT='包含各种类型列的同步表';

现在需要将其同步到另一个 MySQLobj_db

表元数据

表的元信息比较少,包括表名、表类型、表说明(表注释)等,其他的,类似字符集、排序规则等,就继承数据库的

表名,我想你们都知道,也就是对应上面的 tbl_sync

表说明(表注释)你们肯定也知道,对应上面的 包含各种类型列的同步表

那表类型是什么,你们还知道吗?

我们通常说的表是狭义上的表,也就是 基本表,是最常见的表类型,用于存储具有明确定义的列和数据类型的数据

tbl_sync 就是 基本表 ,但广义上的表还包括 视图临时表系统表 等等

下文都是基于 基本表 ,大家需要注意这个前提

通过 jdbc 获取 表元数据,非常简单,直接看代码

Connection connection = dataSource.getConnection();
DatabaseMetaData databaseMetaData = connection.getMetaData();
ResultSet tableResultSet = databaseMetaData.getTables(connection.getCatalog(), connection.getSchema(),
		"tbl_sync", new String[]{"TABLE"});
while (tableResultSet.next()) {
	System.out.println("tableCatalog = " + tableResultSet.getString("TABLE_CAT"));
	System.out.println("tableSchema = " + tableResultSet.getString("TABLE_SCHEM"));
	System.out.println("tableName = " + tableResultSet.getString("TABLE_NAME"));
	System.out.println("tableType = " + tableResultSet.getString("TABLE_TYPE"));
	System.out.println("remarks = " + tableResultSet.getString("REMARKS"));
}

输出结果

tableCatalog = test
tableSchema = null
tableName = tbl_sync
tableType = TABLE
remarks = 包含各种类型列的同步表

一般我们只需要关注: TABLE_NAMETABLE_TYPEREMARKS

我们看下 java.sql.DatabaseMetaData#getTables 说明

点击查看代码
/**
 * Retrieves a description of the tables available in the given catalog.
 * Only table descriptions matching the catalog, schema, table
 * name and type criteria are returned.  They are ordered by
 * <code>TABLE_TYPE</code>, <code>TABLE_CAT</code>,
 * <code>TABLE_SCHEM</code> and <code>TABLE_NAME</code>.
 * <P>
 * Each table description has the following columns:
 *  <OL>
 *  <LI><B>TABLE_CAT</B> String {@code =>} table catalog (may be <code>null</code>)
 *  <LI><B>TABLE_SCHEM</B> String {@code =>} table schema (may be <code>null</code>)
 *  <LI><B>TABLE_NAME</B> String {@code =>} table name
 *  <LI><B>TABLE_TYPE</B> String {@code =>} table type.  Typical types are "TABLE",
 *                  "VIEW", "SYSTEM TABLE", "GLOBAL TEMPORARY",
 *                  "LOCAL TEMPORARY", "ALIAS", "SYNONYM".
 *  <LI><B>REMARKS</B> String {@code =>} explanatory comment on the table
 *  <LI><B>TYPE_CAT</B> String {@code =>} the types catalog (may be <code>null</code>)
 *  <LI><B>TYPE_SCHEM</B> String {@code =>} the types schema (may be <code>null</code>)
 *  <LI><B>TYPE_NAME</B> String {@code =>} type name (may be <code>null</code>)
 *  <LI><B>SELF_REFERENCING_COL_NAME</B> String {@code =>} name of the designated
 *                  "identifier" column of a typed table (may be <code>null</code>)
 *  <LI><B>REF_GENERATION</B> String {@code =>} specifies how values in
 *                  SELF_REFERENCING_COL_NAME are created. Values are
 *                  "SYSTEM", "USER", "DERIVED". (may be <code>null</code>)
 *  </OL>
 *
 * <P><B>Note:</B> Some databases may not return information for
 * all tables.
 *
 * @param catalog a catalog name; must match the catalog name as it
 *        is stored in the database; "" retrieves those without a catalog;
 *        <code>null</code> means that the catalog name should not be used to narrow
 *        the search
 * @param schemaPattern a schema name pattern; must match the schema name
 *        as it is stored in the database; "" retrieves those without a schema;
 *        <code>null</code> means that the schema name should not be used to narrow
 *        the search
 * @param tableNamePattern a table name pattern; must match the
 *        table name as it is stored in the database
 * @param types a list of table types, which must be from the list of table types
 *         returned from {@link #getTableTypes},to include; <code>null</code> returns
 * all types
 * @return <code>ResultSet</code> - each row is a table description
 * @exception SQLException if a database access error occurs
 * @see #getSearchStringEscape
 */

相信你们都能看懂,我只强调下 TABLE_TYPE

其值包括

  • TABLE
  • VIEW
  • SYSTEM TABLE
  • GLOBAL TEMPORARY,LOCAL TEMPORARY
  • ALIAS
  • SYNONYM

列元数据

列元信息比较多一点,包括列名、列类型、列类型名、是否自增、是否允许NULL、列大小、小数位数、默认值、列说明(列注释)等

通过 jdbc 获取 列元数据 也很简单,直接看代码

Connection connection = dataSource.getConnection();
DatabaseMetaData databaseMetaData = connection.getMetaData();
ResultSet columnResultSet = databaseMetaData.getColumns(connection.getCatalog(), connection.getSchema(),
		"tbl_sync", null);
while (columnResultSet.next()) {
	System.out.println("ColumnName = " + columnResultSet.getString("COLUMN_NAME"));
	System.out.println("ColumnType = " + columnResultSet.getInt("DATA_TYPE"));
	System.out.println("ColumnTypeName = " + columnResultSet.getString("TYPE_NAME"));
	System.out.println("isAutoIncrement = " + columnResultSet.getString("IS_AUTOINCREMENT"));
	System.out.println("isNullable = " + columnResultSet.getString("IS_NULLABLE"));
	System.out.println("Precision = " + columnResultSet.getInt("COLUMN_SIZE"));
	System.out.println("Scale = " + columnResultSet.getInt("DECIMAL_DIGITS"));
	System.out.println("DefaultValue = " + columnResultSet.getString("COLUMN_DEF"));
	System.out.println("Remarks = " + columnResultSet.getString("REMARKS"));
	System.out.println("===================================");
}

输出结果

ColumnName = c_bigint_auto
ColumnType = -5
ColumnTypeName = BIGINT UNSIGNED
isAutoIncrement = YES
isNullable = NO
Precision = 20
Scale = 0
DefaultValue = null
Remarks = bigint 类型
===================================
ColumnName = c_bigint
ColumnType = -5
ColumnTypeName = BIGINT
isAutoIncrement = NO
isNullable = YES
Precision = 19
Scale = 0
DefaultValue = null
Remarks = bigint 类型
===================================
ColumnName = c_vachar
ColumnType = 12
ColumnTypeName = VARCHAR
isAutoIncrement = NO
isNullable = NO
Precision = 100
Scale = 0
DefaultValue = null
Remarks = varchar 类型
===================================
...

ColumnType 的值是 java.sql.Types-5 即是 java.sql.Types#BIGINT

java.sql.Types

那是不是根据 ColumnType 就可以推断出数据库列类型了?

我们看下如下输出

ColumnName = c_longtext
ColumnType = -1
ColumnTypeName = LONGTEXT
isAutoIncrement = NO
isNullable = YES
Precision = 2147483647
Scale = 0
DefaultValue = null
Remarks = longtext 类型
===================================
ColumnName = c_json
ColumnType = -1
ColumnTypeName = JSON
isAutoIncrement = NO
isNullable = YES
Precision = 1073741824
Scale = 0
DefaultValue = null
Remarks = json 类型
===================================

ColumnType = -1 是对应 LONGTEXT ,还是对应 JSON

我们再看一个

ColumnName = c_datetime
ColumnType = 93
ColumnTypeName = DATETIME
isAutoIncrement = NO
isNullable = YES
Precision = 23
Scale = 0
DefaultValue = null
Remarks = datetime 类型
===================================
ColumnName = c_timestamp
ColumnType = 93
ColumnTypeName = TIMESTAMP
isAutoIncrement = NO
isNullable = YES
Precision = 24
Scale = 0
DefaultValue = null
Remarks = timestamp 类型
===================================

ColumnType = 93 是对应 DATETIME,还是对应 TIMESTAMP

这说明不能通过 java.sql.Types 精准确认列的数据库类型!!!

那怎么办?

我相信你们已经看到了列的另一个元数据:ColumnTypeName

它不就是 源数据源 中列列类型吗?

比如列 c_timestamp 的类型不就是 TIMESTAMP 吗,丝毫不差,准确的很!

但是我们不能忘了我们的初衷:拼接目标表的建表 SQL

通过 ColumnTypeName 能不能对应到目标表的列类型?

直接使用,肯定是不行的,关系型数据库之间的类型不是完全一一对应的,比如 MySQLDATETIMEOracle 是没有的

那可不可以通过 ColumnTypeName 来映射了,比如 DATETIME 映射到 OracleDATE

理论上来说是可行的,但是,问题又来了!

我们是通过 jdbc 来完成映射的,它只提供了int 类型的 java.sql.Types ,并未提供 String 类型的 java.sql.Types

莫非你要自实现 String 类型的 java.sql.Types ? 你穷举的过来吗?

所以我们需要根据 java.sql.Types 对源数据源的列类型最大兼容性获取,而不是百分之百的精准获取

例如:java.sql.Types#LONGVARCHAR 就当作列类型 LONGTEXT ,然后向目标数据源映射

如果想更精准,则再结合 ColumnTypeName 的值向目标数据源映射

总之一句话:ColumnType 主导,ColumnTypeName 辅助,完成目标数据源列映射

java.sql.DatabaseMetaData#getColumns 能获取的元数据不局限于上述示例中的那些

大家可以去看下其源码注释,因为太长了,我就不贴了,我们重点看下 COLUMN_SIZE

* The COLUMN_SIZE column specifies the column size for the given column.
* For numeric data, this is the maximum precision.  For character data, this is the length in characters.
* For datetime datatypes, this is the length in characters of the String representation (assuming the
* maximum allowed precision of the fractional seconds component). For binary data, this is the length in bytes.  For the ROWID datatype,
* this is the length in bytes. Null is returned for data types where the
* column size is not applicable.

我给你们逐行翻译下

/**
 *  For numeric data, this is the maximum precision => 对于数值数据,表示最大精度
 *  For character data, this is the length in characters => 对于字符数据,表示字符长度
 *  For datetime datatypes, this is the length in characters of the String representation(assuming the maximum allowed precision of the fractional seconds component )
 *      => 对于日期时间数据类型,表示字符串表示形式的最大长度(假设最大允许的分秒小数部分的精度)
 *      例如:"2024-04-30 14:00:00" => 19,"2024-04-30 14:00:00.234" => 23
 *      "14:00:00" => 8,"14:00:00.234" => 11
 *  For binary data, this is the length in bytes => 对于二进制数据,表示字节长度
 *  For the ROWID datatype, this is the length in bytes => 对于 ROWID 类型,表示字节长度
 *  0 is returned for data types where the column size is not applicable => 对于列大小不适用的数据类型,返回0
 */

主键元数据

主键元信息就比较少了,我们一般只关注主键名、列名、列序号

通过 jdbc 代码获取,示例代码如下

Connection connection = dataSource.getConnection();
DatabaseMetaData databaseMetaData = connection.getMetaData();
ResultSet primaryKeysResultSet = databaseMetaData.getPrimaryKeys(connection.getCatalog(), connection.getSchema(), "tbl_sync");
while (primaryKeysResultSet.next()) {
	String columnName = primaryKeysResultSet.getString("COLUMN_NAME");
	short keySeq = primaryKeysResultSet.getShort("KEY_SEQ");
	String pkName = primaryKeysResultSet.getString("PK_NAME");
	System.out.println(columnName + " - " + keySeq + " - " + pkName);
}

输出结果

c_vachar - 1 - PRIMARY
c_char - 2 - PRIMARY
c_bigint_auto - 3 - PRIMARY

不用过多说明了吧,你们肯定都能看懂

索引元数据

与主键元数据类似,关注的元数据主要包括索引名、列名、列序号,同时多了一个 是否非唯一

通过 jdbc 获取,代码如下

Connection connection = dataSource.getConnection();
DatabaseMetaData databaseMetaData = connection.getMetaData();
ResultSet indexResultSet = databaseMetaData.getIndexInfo(connection.getCatalog(), connection.getSchema(), "tbl_sync", false, false);
while (indexResultSet.next()) {
	String indexName = indexResultSet.getString("INDEX_NAME");
	String columnName = indexResultSet.getString("COLUMN_NAME");
	boolean nonUnique = indexResultSet.getBoolean("NON_UNIQUE");
	short ordinalPosition = indexResultSet.getShort("ORDINAL_POSITION");
	System.out.println(columnName + " - " + ordinalPosition + " - " + indexName +  " - " + nonUnique);
}

输出结果

c_vachar - 1 - PRIMARY - false
c_char - 2 - PRIMARY - false
c_bigint_auto - 3 - PRIMARY - false
c_bigint_auto - 1 - uk_id - false
c_vachar - 1 - idx_name_salary - true
c_decimal_4 - 2 - idx_name_salary - true

建表 SQL

当相关元数据都获取到之后,就万事俱备,只欠东风了

我们将 test 库下的表 tbl_sync 同步到另一个 MySQLobj_db

SQL 拼接如下

点击查看代码
public String getCreateTableSql(String schemaName, TableMeta tableMeta, List<ColumnMeta> columnMetas,
								IndexMeta primaryKeyMeta, Map<String, IndexMeta> indexMetaMap) {
	StringBuilder createSql = new StringBuilder("CREATE TABLE " + schemaName + "." + tableMeta.getTableName() + " ( ");
	for (ColumnMeta columnMeta : columnMetas) {
		createSql.append(columnMeta.getColumnName()).append(" ").append(getColumnType(columnMeta));
		if (columnMeta.getIfUnsigned()) {
			createSql.append(" UNSIGNED");
		}
		if (columnMeta.getIfNullable() == 0) {
			createSql.append(" NOT NULL");
		}
		if (StrUtil.isNotBlank(columnMeta.getDefaultValue())) {
			createSql.append(" DEFAULT '").append(columnMeta.getDefaultValue()).append("'");
		}
		if (columnMeta.getIfAutoIncrement()) {
			createSql.append(" AUTO_INCREMENT");
		}
		if (StrUtil.isNotBlank(columnMeta.getRemarks())) {
			createSql.append(" COMMENT '").append(columnMeta.getRemarks()).append("'");
		}
		createSql.append(",");
	}
	// 主键处理
	if (ObjectUtil.isNotNull(primaryKeyMeta)) {
		List<IndexColumnMeta> indexColumns = primaryKeyMeta.getIndexColumns();
		indexColumns.sort(Comparator.comparingInt(IndexColumnMeta::getOrdinalPosition));
		createSql.append(" PRIMARY KEY (");
		for (int i=0; i<indexColumns.size(); i++) {
			if (i>0) {
				createSql.append(",");
			}
			createSql.append(indexColumns.get(i).getColumnName());
		}
		createSql.append("),");
	}
	if (CollectionUtil.isNotEmpty(indexMetaMap)) {
		for (IndexMeta indexMeta : indexMetaMap.values()) {
			if (indexMeta.getIndexType() == IndexTypeEnum.UNIQUE) {
				// 唯一索引
				createSql.append("UNIQUE ");
			}
			createSql.append("KEY ").append(indexMeta.getIndexName()).append(" (");
			List<IndexColumnMeta> indexColumns = indexMeta.getIndexColumns();
			indexColumns.sort(Comparator.comparingInt(IndexColumnMeta::getOrdinalPosition));
			for (int i=0; i<indexColumns.size(); i++) {
				if (i>0) {
					createSql.append(",");
				}
				createSql.append(indexColumns.get(i).getColumnName());
			}
			createSql.append("),");
		}
	}
	// 删除最后一个逗号
	createSql.deleteCharAt(createSql.length()-1);
	createSql.append(")");
	if (StrUtil.isNotBlank(tableMeta.getRemarks())) {
		createSql.append(" COMMENT '").append(tableMeta.getRemarks()).append("'");
	}
	return createSql.toString();
}

/**
 * 获取表 列类型
 * @param columnMeta 列元数据
 * @return mysql 列类型
 */
private String getColumnType(ColumnMeta columnMeta) {
	switch (columnMeta.getColumnType()) {
		// 数值类型
		case Types.TINYINT:
			return "TINYINT";
		case Types.SMALLINT:
			return "SMALLINT";
		case Types.INTEGER:
			return "INT";
		case Types.BIGINT:
			return "BIGINT";
		case Types.FLOAT:
		case Types.REAL:
			return columnMeta.getPrecision() > 0 ? "FLOAT(" + columnMeta.getPrecision() + "," + columnMeta.getScale() + ")" : "FLOAT";
		case Types.DOUBLE:
			return columnMeta.getPrecision() > 0 ? "DOUBLE(" + columnMeta.getPrecision() + "," + columnMeta.getScale() + ")" : "DOUBLE";
		case Types.DECIMAL:
			return "DECIMAL(" + columnMeta.getPrecision() + "," + columnMeta.getScale() + ")";
		case Types.NUMERIC:
			return columnMeta.getScale() <= 0 ? "BIGINT" : "DECIMAL(" + columnMeta.getPrecision() + "," + columnMeta.getScale() + ")";
		// 字符与字符串类型
		case Types.CHAR:
		case Types.NCHAR:
			return columnMeta.getPrecision() > 0 ? "CHAR(" + columnMeta.getPrecision() + ")" : "CHAR";
		case Types.VARCHAR:
		case Types.NVARCHAR:
			return columnMeta.getPrecision() > 0 ? "VARCHAR(" + columnMeta.getPrecision() + ")" : "VARCHAR";
		case Types.LONGVARCHAR:
		case Types.LONGNVARCHAR:
			switch (columnMeta.getColumnTypeName()) {
				case "TINYTEXT":
					return "TINYTEXT";
				case "MEDIUMTEXT":
					return "MEDIUMTEXT";
				case "LONGTEXT":
					return "LONGTEXT";
				case "JSON":
					return "JSON";
				default:
					return "TEXT";
			}
		case Types.CLOB:
		case Types.NCLOB:
			return "LONGTEXT";
		// 日期和时间类型
		case Types.DATE:
			switch (columnMeta.getColumnTypeName()) {
				case "YEAR":
					return "YEAR";
				default:
					return "DATE";
			}
		case Types.TIME:
			return "TIME" + (columnMeta.getPrecision() > 8 ? "(" + (columnMeta.getPrecision() - 9) + ")" : "");
		case Types.TIMESTAMP:
			switch (columnMeta.getColumnTypeName()) {
				case "DATETIME":
					return "DATETIME" + (columnMeta.getPrecision() > 19 ? "(" + (columnMeta.getPrecision() - 20) + ")" : "");
				case "DATE":
					// oracle 的 DATE
					return "DATETIME";
				default:
					return "TIMESTAMP"+ (columnMeta.getPrecision() > 19 ? "(" + (columnMeta.getPrecision() - 20) + ")" : "");
			}
		// 二进制类型
		case Types.BIT:
		case Types.BOOLEAN:
			return columnMeta.getPrecision() > 0 ? "BIT(" + columnMeta.getPrecision() + ")" : "BIT";
		case Types.BINARY:
			return columnMeta.getPrecision() > 0 ? "BINARY(" + columnMeta.getPrecision() + ")" : "BINARY";
		case Types.VARBINARY:
			return columnMeta.getPrecision() > 0 ? "VARBINARY(" + columnMeta.getPrecision() + ")" : "VARBINARY";
		case Types.BLOB:
		case Types.LONGVARBINARY:
			switch (columnMeta.getColumnTypeName()) {
				case "TINYBLOB":
					return "TINYBLOB";
				case "MEDIUMBLOB":
					return "MEDIUMBLOB";
				case "LONGBLOB":
					return "LONGBLOB";
				default:
					return "BLOB";
			}
		case Types.OTHER:
			if (columnMeta.getColumnTypeName().contains("VARCHAR")) {
				return "VARCHAR" + (columnMeta.getPrecision() > 0 ? "(" + columnMeta.getPrecision() + ")" : "");
			} else if (columnMeta.getColumnTypeName().contains("TIMESTAMP")) {
				return "TIMESTAMP" + (columnMeta.getScale() > 0 ? "(" + columnMeta.getScale() + ")" : "");
			}
			else {
				throw new SyncException("不支持的类型:" + columnMeta.getColumnTypeName());
			}
		default:
			throw new SyncException("不支持的类型:" + columnMeta.getColumnTypeName());
	}
}

结合元数据的获取

点击查看代码
@Test
public void getMySQLCreateTableSql() throws SQLException {
	Connection connection = dataSource.getConnection();
	DatabaseMetaData databaseMetaData = connection.getMetaData();
	ResultSet tableResultSet = databaseMetaData.getTables(connection.getCatalog(), connection.getSchema(), "tbl_sync", new String[]{"TABLE"});
	TableMeta tableMeta = new TableMeta();
	while (tableResultSet.next()) {
		tableMeta.setTableName(tableResultSet.getString("TABLE_NAME"));
		tableMeta.setTableType(tableResultSet.getString("TABLE_TYPE"));
		tableMeta.setRemarks(tableResultSet.getString("REMARKS"));
	}
	// 获取列元数据
	ResultSet columnResultSet = databaseMetaData.getColumns(connection.getCatalog(), connection.getSchema(), "tbl_sync", null);
	List<ColumnMeta> columnMetas = new ArrayList<>();
	while (columnResultSet.next()) {
		ColumnMeta columnMeta = new ColumnMeta();
		columnMeta.setColumnName(columnResultSet.getString("COLUMN_NAME"));
		columnMeta.setColumnType(columnResultSet.getInt("DATA_TYPE"));
		columnMeta.setColumnTypeName(columnResultSet.getString("TYPE_NAME"));
		columnMeta.setIfAutoIncrement("YES".equalsIgnoreCase(columnResultSet.getString("IS_AUTOINCREMENT")));
		columnMeta.setIfNullable("YES".equalsIgnoreCase(columnResultSet.getString("IS_NULLABLE")) ? 1 : 0);
		columnMeta.setPrecision(columnResultSet.getInt("COLUMN_SIZE"));
		columnMeta.setScale(columnResultSet.getInt("DECIMAL_DIGITS"));
		columnMeta.setDefaultValue(columnResultSet.getString("COLUMN_DEF"));
		columnMeta.setRemarks(columnResultSet.getString("REMARKS"));
		columnMeta.setIfUnsigned(columnMeta.getColumnTypeName().contains("UNSIGNED"));
		columnMetas.add(columnMeta);
	}
	columnResultSet.close();
	// 获取主键元数据
	ResultSet primaryKeyResultSet = databaseMetaData.getPrimaryKeys(connection.getCatalog(), connection.getSchema(), "tbl_sync");
	IndexMeta primaryKeyMeta = new IndexMeta();
	while (primaryKeyResultSet.next()) {
		IndexColumnMeta indexColumnMeta = new IndexColumnMeta(primaryKeyResultSet.getString("COLUMN_NAME"), primaryKeyResultSet.getShort("KEY_SEQ"));
		primaryKeyMeta.setIndexName(primaryKeyResultSet.getString("PK_NAME"));
		primaryKeyMeta.getIndexColumns().add(indexColumnMeta);
	}
	primaryKeyResultSet.close();
	// 获取索引元数据
	ResultSet indexResultSet = databaseMetaData.getIndexInfo(connection.getCatalog(), connection.getSchema(), "tbl_sync", false, false);
	Map<String, IndexMeta> indexMetaMap = new HashMap<>();
	while (indexResultSet.next()) {
		String indexName = indexResultSet.getString("INDEX_NAME");
		if (indexName.equals(primaryKeyMeta.getIndexName())) {
			continue;
		}
		IndexMeta indexMeta = indexMetaMap.get(indexName);
		if (ObjectUtil.isNull(indexMeta)) {
			indexMeta = new IndexMeta(indexName);
			indexMetaMap.put(indexName, indexMeta);
		}
		indexMeta.setIndexType(indexResultSet.getBoolean("NON_UNIQUE") ? IndexTypeEnum.NORMAL : IndexTypeEnum.UNIQUE);
		indexMeta.getIndexColumns().add(new IndexColumnMeta(indexResultSet.getString("COLUMN_NAME"), indexResultSet.getShort("ORDINAL_POSITION")));
	}
	indexResultSet.close();

	MysqlSql mysqlSql = new MysqlSql();
	String createTableSql = mysqlSql.getCreateTableSql("obj_db", tableMeta, columnMetas, primaryKeyMeta, indexMetaMap);
	System.out.println(SQLUtils.formatMySql(createTableSql));
}

得到的建表 SQL 如下

CREATE TABLE obj_db.tbl_sync (
	c_bigint_auto BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'bigint 类型',
	c_bigint BIGINT COMMENT 'bigint 类型',
	c_vachar VARCHAR(100) NOT NULL COMMENT 'varchar 类型',
	c_char CHAR(32) NOT NULL COMMENT 'char 类型',
	c_text TEXT NOT NULL COMMENT 'text 类型',
	c_decimal_4 DECIMAL(15, 4) NOT NULL DEFAULT '5000.0000' COMMENT 'decimal 类型',
	c_decimal_0 DECIMAL(10, 0) COMMENT 'decimal 类型',
	c_blob BLOB COMMENT 'blob 类型',
	c_bit BIT(1) COMMENT 'bit 类型',
	c_tinyint TINYINT COMMENT 'tinyint 类型',
	c_binary BINARY(10) COMMENT 'binary 类型',
	c_float FLOAT(13, 0) COMMENT 'float 类型',
	c_double DOUBLE(23, 0) COMMENT 'double 类型',
	c_varbinary VARBINARY(20) COMMENT 'varbinary 类型',
	c_longblob LONGBLOB COMMENT 'longblob 类型',
	c_longtext LONGTEXT COMMENT 'longtext 类型',
	c_json JSON COMMENT 'json 类型',
	c_date DATE COMMENT 'date 类型',
	c_time TIME(2) COMMENT 'time 类型',
	c_datetime DATETIME(3) COMMENT 'datetime 类型',
	c_timestamp TIMESTAMP(4) COMMENT 'timestamp 类型',
	c_year YEAR COMMENT 'year 类型',
	PRIMARY KEY (c_vachar, c_char, c_bigint_auto),
	UNIQUE KEY uk_id (c_bigint_auto),
	KEY idx_name_salary (c_vachar, c_decimal_4)
) COMMENT '包含各种类型列的同步表'

可以看出,与原表的结构是一致的!

此处应该有掌声

同源同步

何谓同源?

就是数据库类型相同的数据源,例如从 MySQL 同步到 MySQL

这种情况还有必要进行 SQL 拼接吗?

还记得怎么查看 MySQL 表的完整定义吗

SHOW CREATE TABLE test.tbl_sync

这是不是就可以获取到表的 DDL

所以同源的表结构同步,就不用拼接 SQL 那么复杂了,直接获取 DDL 后在目标数据源建表即可

总结

  • 异构数据源同步的策略有两种:离线同步 和 实时同步,各自的特点及使用场景需要区分清楚
  • 关系型数据库的元数据有很多种,大家可以仔细看看 java.sql.DatabaseMetaData
  • 同源表结构同步,可以不用拼接建表 SQL,可以直接获取建表 DDL
  • 异源表结构同步,需要先获取源表的相关元数据,然后再拼接目标表的建表 SQL,最后在目标数据源执行 SQL 创建目标表
  • COLUMN_SIZE 针对不同的列类型,它的含义不同,文中已经详细说明,值得大家注意

标签:COMMENT,case,同步,return,columnMeta,数据源,之表,类型,Types
From: https://www.cnblogs.com/youzhibing/p/18172471

相关文章

  • [转]ptp(precision time protocol)时钟同步
    一、介绍1:什么是ptpPTP(PrecisionTimeProtocol)是一个通过网络同步时钟的一个协议。当硬件支持时,PTP精度能达到亚微秒,比NTP(NetworkTimeProtocol)精度更高。2:ptp应用场景1)数据中心数据中心需要NTP/PTP同步,以确保集群的时域运行。同步对于虚拟机计算是必不可少的。日志事件的......
  • Qt/C++音视频开发72-倍速推流/音视频同步倍速推流/不改变帧率和采样率/低倍速和高倍速
    一、前言最近多了个新需求,需要倍速推流,推流界的扛把子obs也有倍速推流功能,最高支持到两倍速。这里所说的倍速,当然只限定在文件,只有文件才可能有倍速功能,因为也只有文件才能倍速解码播放。实时视频流是不可能倍速的,因为没有时长,有时长的才可以按照播放进度来。是否是文件也不能通......
  • 同步与互斥的失败例子-02
    一句话理解同步与互斥:我等你用完厕所,我再用厕所。什么叫同步?就是条件不允许,我要等等。什么是互斥?你我早起都要用厕所,谁先抢到谁先用,中途不被打扰。同步与互斥经常放在一起讲,是因为它们之的关系很大,“互斥”操作可以使用“同步”来实现。我“等”你用完厕所,我再用厕所。这不就是......
  • CentOS7的Hadoop集群配置时间同步
    mastervi/etc/ntp.conf点击查看代码[root@masterhadoop]#vim/etc/ntp.conf...前文省略..#Hostsonlocalnetworkarelessrestricted.restrict192.168.10.0mask255.255.255.0nomodifynotrap#Usepublicserversfromthepool.ntp.orgproject.#Please......
  • 好用的自动同步软件:FreeFileSync
    最近研究怎么能够将文件从电脑上自动备份到移动硬盘,发现了很多人都在安利的FreeFileSync(官网:FreeFileSync )FreeFileSync​freefilesync.org/这是一个免费软件,在各个系统中都有提供安装包。软件安装过程非常简单,目前我将其直接安装到了C盘,安装好后即可看到两个软件:绿色的FreeFil......
  • 保险公司数据中心同步数据方案,该怎么选才既合规又易于管理?
    保险公司数据中心是一种集中存储、处理和管理保险公司数据的设施。它由一系列硬件和软件组成,包括服务器、存储设备、网络设备等,以提供一个高效、安全和可靠的环境来处理和存储大量数据。 在保险公司的运营中,数据中心扮演着至关重要的角色。它支持保险公司的核心业务系统,如保单......
  • 面临文件同步需求时 大文件同步方案要怎么选择?
    大文件同步在企业数据管理中是一个常见的需求,但在实际操作过程中可能会遇到一系列问题,导致业务效率降低、管理困难。 面临的主要问题包括:1、传输速度慢:大文件需要较长时间来传输,尤其是在网络带宽有限的情况下,传输效率会更低。2、断点续传问题:在不稳定的网络环境中,大文件传输......
  • [转]解决Win7和Linux Deepin双系统时间不同步的问题[2017.3.13 sina blog]
    原博地址:http://xsinger.me/diy/261.html/comment-page-1对于双系统的用户,有时候从Linux回到Windows的时候,时间总相差8小时。为什么LinuxDeepin和Windows双系统会有时间差因为安装LinuxDeepin时选择了UTC(协调世界时)时间,所以LinuxDeepin开机总是从互联网获取时间并且写入BIOS......
  • Elasticsearch 系列(六)- ES数据同步和ES集群
    本章将和大家分享ES的数据同步方案和ES集群相关知识。废话不多说,下面我们直接进入主题。一、ES数据同步1、数据同步问题Elasticsearch中的酒店数据来自于mysql数据库,因此mysql数据发生改变时,Elasticsearch也必须跟着改变,这个就是Elasticsearch与mysql之间的数据同步。在微服务......
  • 同步产品 到 A8 V3 的注意项
    1、如果同步到erp的产品,在erp中 新增【生产拆装单】 的时候,选择产品,点击选择之后,提示报错的情况  可以执行下面的 sql在 db去执行,批量处理UPDATE ptype SET CUSTOMITEMID=NULL,CUSTOMITEMFROM=NULL,CUSTOMITEMNAME=NULL,CUSTOMITEMREFID=NULL为了避免有些版本不......