1. 背景在 ETLExtract-Transform-Load或数据集成平台中源端数据库的表结构、字段类型、索引、主键、外键等元数据信息是数据抽取与映射的基础。传统方案普遍通过直接查询各数据库的系统表来获取这些信息例如数据库表信息字段信息MySQLinformation_schema.TABLESinformation_schema.COLUMNSOracledba_tables/all_tablesdba_tab_columnsSQL Serversys.tablessys.columnsPostgreSQLinformation_schema.tablesinformation_schema.columns这种方式存在一个明显弊端每接入一种新数据库就需要编写一套针对该系统表的 SQL 查询逻辑维护成本随数据源种类呈线性增长。2. 需求构建一套统一的元数据同步机制满足以下要求一次编写多库通用— 不依赖各数据库的系统表差异支持动态数据源切换— 可在运行时按数据源名称获取元数据异步非阻塞同步— 元数据同步过程不阻塞主线程增量更新— 只处理变更部分删除已不存在的旧元数据多租户隔离— 元数据按租户维度存储3. 设计思路3.1 核心原理JDBC 规范提供了java.sql.DatabaseMetaData接口该接口定义了一系列获取数据库元数据的标准方法。所有符合 JDBC 标准的数据库驱动都必须实现此接口因此无需关心底层是 MySQL、Oracle 还是 Hive。┌──────────────────────────────────────────────────────┐ │ qrcb-meta 服务 │ │ │ │ MetaSchemaService MetaTableService │ │ MetaColumnService MetaIndexService │ │ MetaPrimaryKeyService MetaForeignKeyService │ │ │ │ ↓ 统一入口: java.sql.DatabaseMetaData │ │ │ │ ┌──────────┬──────────┬──────────┬──────────┐ │ │ │ MySQL │ Oracle │ Hive │ 其他JDBC │ │ │ │ Driver │ Driver │ Driver │ Driver │ │ │ └──────────┴──────────┴──────────┴──────────┘ │ └──────────────────────────────────────────────────────┘3.2 元数据层级元数据按三层结构组织层级对应 API实体说明1. SchemametaData.getSchemas()MetaSchema数据库 / 模式2. TablemetaData.getTables()MetaTable表 / 视图3. Column/Index/PK/FKmetaData.getColumns()等MetaColumn等字段、索引、约束3.3 同步流程触发同步 (asyncSchemaMetadata / asyncTableMetadata) │ ├─ 1. 通过 DynamicDataSourceGenResolver 获取目标 DataSource │ ├─ 2. 获取连接 → connection.getMetaData() │ ├─ 3. 调用 DatabaseMetaData 方法获取实时元数据 │ ├─ getSchemas() → ListMetaSchema │ ├─ getTables() → ListMetaTable │ ├─ getColumns() → ListMetaColumn │ ├─ getIndexInfo() → ListMetaIndex │ ├─ getPrimaryKeys()→ ListMetaPrimaryKey │ └─ getImportedKeys()→ ListMetaForeignKey │ ├─ 4. 对比已存储的元数据删除不存在的旧记录 │ ├─ 5. 设置关联主键 (schemaId, tableId)批量 upsert │ └─ 6. 异步执行 (ThreadPoolTaskExecutor TenantBroker)4. 核心依赖包依赖用途java.sql.DatabaseMetaDataJDK 内置元数据查询核心接口java.sql.ConnectionJDK 内置数据库连接javax.sql.DataSourceJDK 内置数据源抽象com.qrcb.common.core.datasource项目内部动态数据源解析 (DynamicDataSourceGenResolver)com.qrcb.common.core.assemble.util.JdbcUtils项目内部JDBC ResultSet 安全读取工具com.qrcb.common.core.data.tenant.TenantBroker项目内部租户上下文代理cn.hutool.core.util.StrUtilHutool字符串工具com.baomidou.mybatisplus.extension.service.impl.ServiceImplMyBatis-Plus持久化基类org.springframework.scheduling.concurrent.ThreadPoolTaskExecutorSpring异步线程池5. 核心代码片段5.1 动态获取表列表Schema 级// MetaTableServiceImpl.getTablesDynamic() public ListMetaTable getTablesDynamic(String catalog, String schema, String table, String dsName) { DataSource dataSource DynamicDataSourceGenResolver.fetch(dsName); ListMetaTable tableList new ArrayList(); try (Connection conn dataSource.getConnection()) { DatabaseMetaData metaData conn.getMetaData(); // ★ 核心 API获取所有表 try (ResultSet rs metaData.getTables(catalog, schema, table, null)) { ResultSetMetaData rsMeta rs.getMetaData(); while (rs.next()) { int i 1; MetaTable t new MetaTable(); t.setCatalogName(JdbcUtils.getStringMetaData(rsMeta, rs, i)); t.setSchemaName(JdbcUtils.getStringMetaData(rsMeta, rs, i)); t.setTableName(JdbcUtils.getStringMetaData(rsMeta, rs, i)); t.setTableType(JdbcUtils.getStringMetaData(rsMeta, rs, i)); t.setRemarks(JdbcUtils.getStringMetaData(rsMeta, rs, i)); // ... 其余元数据字段 tableList.add(t); } } } return filterTable(tableList, catalog, schema, table); }5.2 获取字段信息// MetaColumnServiceImpl.getColumnsDynamic() public ListMetaColumn getColumnsDynamic(String catalog, String schema, String table, String column, String dsName) { DataSource dataSource DynamicDataSourceGenResolver.fetch(dsName); ListMetaColumn columnList new ArrayList(); try (Connection conn dataSource.getConnection()) { DatabaseMetaData metaData conn.getMetaData(); // ★ 核心 API获取表字段 try (ResultSet rs metaData.getColumns(catalog, schema, table, column)) { ResultSetMetaData rsMeta rs.getMetaData(); while (rs.next()) { int i 1; MetaColumn col new MetaColumn(); col.setColumnName(JdbcUtils.getStringMetaData(rsMeta, rs, i)); col.setDataType(JdbcUtils.getIntMetaData(rsMeta, rs, i)); col.setTypeName(JdbcUtils.getStringMetaData(rsMeta, rs, i)); col.setColumnSize(JdbcUtils.getIntMetaData(rsMeta, rs, i)); col.setDecimalDigits(JdbcUtils.getIntMetaData(rsMeta, rs, i)); col.setNullable(JdbcUtils.getIntMetaData(rsMeta, rs, i)); col.setRemarks(JdbcUtils.getStringMetaData(rsMeta, rs, i)); col.setColumnDefault(JdbcUtils.getStringMetaData(rsMeta, rs, i)); col.setOrdinalPos(JdbcUtils.getIntMetaData(rsMeta, rs, i)); col.setIsAutoInc(JdbcUtils.getStringMetaData(rsMeta, rs, i)); // ★ 完整映射 23 个 DatabaseMetaData.getColumns 返回列 columnList.add(col); } } } return columnList; }5.3 Schema 获取含 MySQL 兼容处理// MetaSchemaServiceImpl.getSchemasDynamic() public ListMetaSchema getSchemasDynamic(String catalog, String schema, String dsName) { DataSource dataSource DynamicDataSourceGenResolver.fetch(dsName); ListMetaSchema schemaList new ArrayList(); try (Connection conn dataSource.getConnection()) { DatabaseMetaData metaData conn.getMetaData(); // ★ 标准方式获取 Schema try (ResultSet rs metaData.getSchemas(catalog, schema)) { while (rs.next()) { MetaSchema s new MetaSchema(); s.setSchemaName(JdbcUtils.getStringMetaData(rs.getMetaData(), rs, 1)); s.setCatalogName(JdbcUtils.getStringMetaData(rs.getMetaData(), rs, 2)); schemaList.add(s); } } // ★ MySQL 兼容getSchemas 可能返回空回退到 getCatalogs if (CollUtil.isEmpty(schemaList) MySQL.equalsIgnoreCase(metaData.getDatabaseProductName())) { try (ResultSet rs metaData.getCatalogs()) { while (rs.next()) { MetaSchema s new MetaSchema(); s.setCatalogName(JdbcUtils.getStringMetaData(rs.getMetaData(), rs, 1)); schemaList.add(s); } } } } return filterSchema(schemaList, catalog, schema); }5.4 异步同步入口多租户 线程池// MetaTableServiceImpl.asyncTableMetadata() public Boolean asyncTableMetadata(String catalog, String schema, String table, String dsName) { Long tenantId TenantContextHolder.getTenantId(); // ★ 异步提交到线程池同时保留租户上下文 threadPoolTaskExecutor.submit(() - TenantBroker.runAs(tenantId, (id) - this.syncTableMetadata(crrSchema, table, dsName)) ); return true; }5.5 完整的同步事务// MetaTableServiceImpl.syncTableMetadata() Transactional(rollbackFor Exception.class) public synchronized Boolean syncTableMetadata(MetaSchema schema, String tableName, String dsName) { // 1. 同步表基本信息 ListMetaTable tables getTablesDynamic(catalog, schema, tableName, dsName); // 删除不存在的表 deleteTableNotExist(tableIdList); // upsert 表信息 saveOrUpdateBatch(tables); for (MetaTable table : tables) { // 2. 同步字段 ListMetaColumn columns metaColumnService.getColumnsDynamic(...); // 删除不存在的字段 → upsert 字段 // 3. 同步索引 ListMetaIndex indexes metaIndexService.getIndexInfo(...); // 4. 同步主键 ListMetaPrimaryKey pks metaPrimaryKeyService.getPrimaryKeys(...); // 5. 同步外键 ListMetaForeignKey fks metaForeignKeyService.getImportedKeys(...); } return true; }6. DatabaseMetaData 核心 API 对照表JDBC API返回用途getSchemas(catalog, schema)ResultSet获取数据库 Schema 列表getCatalogs()ResultSet获取 Catalog 列表MySQL 回退getTables(catalog, schema, table, types)ResultSet获取表/视图列表10 列getColumns(catalog, schema, table, column)ResultSet获取字段详情23 列getIndexInfo(catalog, schema, table, unique, approximate)ResultSet获取索引信息getPrimaryKeys(catalog, schema, table)ResultSet获取主键列getImportedKeys(catalog, schema, table)ResultSet获取外键列getDatabaseProductName()String获取数据库产品名getDatabaseProductVersion()String获取数据库版本7. 方案优势对比维度系统表方案JDBC DatabaseMetaData 方案兼容性每种数据库需单独编写 SQL一次编写所有 JDBC 驱动通用维护成本随数据库种类线性增长恒定类型映射需自行维护类型映射表getColumns()返回标准java.sql.Types新数据库接入需研究其系统表结构只需配置 JDBC 连接标准合规依赖数据库厂商实现遵循 JDBC 规范版本升级系统表结构可能变化JDBC 驱动保证向后兼容8. 注意事项MySQL 特殊处理MySQL 的getSchemas()可能返回空需回退到getCatalogs()空值安全ResultSet.getXXX()在值为NULL时会抛异常应使用JdbcUtils.getXXXMetaData()安全读取性能考量大表量场景建议按 Schema 分批同步避免一次性加载过多元数据连接管理务必使用 try-with-resources 确保 Connection / ResultSet 正确关闭租户隔离异步同步时通过TenantBroker.runAs()保证租户上下文在线程池中正确传递幂等设计通过Idempotent注解防止短时间内重复提交同步任务
1. 背景在 ETLExtract-Transform-Load或数据集成平台中源端数据库的表结构、字段类型、索引、主键、外键等元数据信息是数据抽取与映射的基础。传统方案普遍通过直接查询各数据库的系统表来获取这些信息例如数据库表信息字段信息MySQLinformation_schema.TABLESinformation_schema.COLUMNSOracledba_tables/all_tablesdba_tab_columnsSQL Serversys.tablessys.columnsPostgreSQLinformation_schema.tablesinformation_schema.columns这种方式存在一个明显弊端每接入一种新数据库就需要编写一套针对该系统表的 SQL 查询逻辑维护成本随数据源种类呈线性增长。2. 需求构建一套统一的元数据同步机制满足以下要求一次编写多库通用— 不依赖各数据库的系统表差异支持动态数据源切换— 可在运行时按数据源名称获取元数据异步非阻塞同步— 元数据同步过程不阻塞主线程增量更新— 只处理变更部分删除已不存在的旧元数据多租户隔离— 元数据按租户维度存储3. 设计思路3.1 核心原理JDBC 规范提供了java.sql.DatabaseMetaData接口该接口定义了一系列获取数据库元数据的标准方法。所有符合 JDBC 标准的数据库驱动都必须实现此接口因此无需关心底层是 MySQL、Oracle 还是 Hive。┌──────────────────────────────────────────────────────┐ │ qrcb-meta 服务 │ │ │ │ MetaSchemaService MetaTableService │ │ MetaColumnService MetaIndexService │ │ MetaPrimaryKeyService MetaForeignKeyService │ │ │ │ ↓ 统一入口: java.sql.DatabaseMetaData │ │ │ │ ┌──────────┬──────────┬──────────┬──────────┐ │ │ │ MySQL │ Oracle │ Hive │ 其他JDBC │ │ │ │ Driver │ Driver │ Driver │ Driver │ │ │ └──────────┴──────────┴──────────┴──────────┘ │ └──────────────────────────────────────────────────────┘3.2 元数据层级元数据按三层结构组织层级对应 API实体说明1. SchemametaData.getSchemas()MetaSchema数据库 / 模式2. TablemetaData.getTables()MetaTable表 / 视图3. Column/Index/PK/FKmetaData.getColumns()等MetaColumn等字段、索引、约束3.3 同步流程触发同步 (asyncSchemaMetadata / asyncTableMetadata) │ ├─ 1. 通过 DynamicDataSourceGenResolver 获取目标 DataSource │ ├─ 2. 获取连接 → connection.getMetaData() │ ├─ 3. 调用 DatabaseMetaData 方法获取实时元数据 │ ├─ getSchemas() → ListMetaSchema │ ├─ getTables() → ListMetaTable │ ├─ getColumns() → ListMetaColumn │ ├─ getIndexInfo() → ListMetaIndex │ ├─ getPrimaryKeys()→ ListMetaPrimaryKey │ └─ getImportedKeys()→ ListMetaForeignKey │ ├─ 4. 对比已存储的元数据删除不存在的旧记录 │ ├─ 5. 设置关联主键 (schemaId, tableId)批量 upsert │ └─ 6. 异步执行 (ThreadPoolTaskExecutor TenantBroker)4. 核心依赖包依赖用途java.sql.DatabaseMetaDataJDK 内置元数据查询核心接口java.sql.ConnectionJDK 内置数据库连接javax.sql.DataSourceJDK 内置数据源抽象com.qrcb.common.core.datasource项目内部动态数据源解析 (DynamicDataSourceGenResolver)com.qrcb.common.core.assemble.util.JdbcUtils项目内部JDBC ResultSet 安全读取工具com.qrcb.common.core.data.tenant.TenantBroker项目内部租户上下文代理cn.hutool.core.util.StrUtilHutool字符串工具com.baomidou.mybatisplus.extension.service.impl.ServiceImplMyBatis-Plus持久化基类org.springframework.scheduling.concurrent.ThreadPoolTaskExecutorSpring异步线程池5. 核心代码片段5.1 动态获取表列表Schema 级// MetaTableServiceImpl.getTablesDynamic() public ListMetaTable getTablesDynamic(String catalog, String schema, String table, String dsName) { DataSource dataSource DynamicDataSourceGenResolver.fetch(dsName); ListMetaTable tableList new ArrayList(); try (Connection conn dataSource.getConnection()) { DatabaseMetaData metaData conn.getMetaData(); // ★ 核心 API获取所有表 try (ResultSet rs metaData.getTables(catalog, schema, table, null)) { ResultSetMetaData rsMeta rs.getMetaData(); while (rs.next()) { int i 1; MetaTable t new MetaTable(); t.setCatalogName(JdbcUtils.getStringMetaData(rsMeta, rs, i)); t.setSchemaName(JdbcUtils.getStringMetaData(rsMeta, rs, i)); t.setTableName(JdbcUtils.getStringMetaData(rsMeta, rs, i)); t.setTableType(JdbcUtils.getStringMetaData(rsMeta, rs, i)); t.setRemarks(JdbcUtils.getStringMetaData(rsMeta, rs, i)); // ... 其余元数据字段 tableList.add(t); } } } return filterTable(tableList, catalog, schema, table); }5.2 获取字段信息// MetaColumnServiceImpl.getColumnsDynamic() public ListMetaColumn getColumnsDynamic(String catalog, String schema, String table, String column, String dsName) { DataSource dataSource DynamicDataSourceGenResolver.fetch(dsName); ListMetaColumn columnList new ArrayList(); try (Connection conn dataSource.getConnection()) { DatabaseMetaData metaData conn.getMetaData(); // ★ 核心 API获取表字段 try (ResultSet rs metaData.getColumns(catalog, schema, table, column)) { ResultSetMetaData rsMeta rs.getMetaData(); while (rs.next()) { int i 1; MetaColumn col new MetaColumn(); col.setColumnName(JdbcUtils.getStringMetaData(rsMeta, rs, i)); col.setDataType(JdbcUtils.getIntMetaData(rsMeta, rs, i)); col.setTypeName(JdbcUtils.getStringMetaData(rsMeta, rs, i)); col.setColumnSize(JdbcUtils.getIntMetaData(rsMeta, rs, i)); col.setDecimalDigits(JdbcUtils.getIntMetaData(rsMeta, rs, i)); col.setNullable(JdbcUtils.getIntMetaData(rsMeta, rs, i)); col.setRemarks(JdbcUtils.getStringMetaData(rsMeta, rs, i)); col.setColumnDefault(JdbcUtils.getStringMetaData(rsMeta, rs, i)); col.setOrdinalPos(JdbcUtils.getIntMetaData(rsMeta, rs, i)); col.setIsAutoInc(JdbcUtils.getStringMetaData(rsMeta, rs, i)); // ★ 完整映射 23 个 DatabaseMetaData.getColumns 返回列 columnList.add(col); } } } return columnList; }5.3 Schema 获取含 MySQL 兼容处理// MetaSchemaServiceImpl.getSchemasDynamic() public ListMetaSchema getSchemasDynamic(String catalog, String schema, String dsName) { DataSource dataSource DynamicDataSourceGenResolver.fetch(dsName); ListMetaSchema schemaList new ArrayList(); try (Connection conn dataSource.getConnection()) { DatabaseMetaData metaData conn.getMetaData(); // ★ 标准方式获取 Schema try (ResultSet rs metaData.getSchemas(catalog, schema)) { while (rs.next()) { MetaSchema s new MetaSchema(); s.setSchemaName(JdbcUtils.getStringMetaData(rs.getMetaData(), rs, 1)); s.setCatalogName(JdbcUtils.getStringMetaData(rs.getMetaData(), rs, 2)); schemaList.add(s); } } // ★ MySQL 兼容getSchemas 可能返回空回退到 getCatalogs if (CollUtil.isEmpty(schemaList) MySQL.equalsIgnoreCase(metaData.getDatabaseProductName())) { try (ResultSet rs metaData.getCatalogs()) { while (rs.next()) { MetaSchema s new MetaSchema(); s.setCatalogName(JdbcUtils.getStringMetaData(rs.getMetaData(), rs, 1)); schemaList.add(s); } } } } return filterSchema(schemaList, catalog, schema); }5.4 异步同步入口多租户 线程池// MetaTableServiceImpl.asyncTableMetadata() public Boolean asyncTableMetadata(String catalog, String schema, String table, String dsName) { Long tenantId TenantContextHolder.getTenantId(); // ★ 异步提交到线程池同时保留租户上下文 threadPoolTaskExecutor.submit(() - TenantBroker.runAs(tenantId, (id) - this.syncTableMetadata(crrSchema, table, dsName)) ); return true; }5.5 完整的同步事务// MetaTableServiceImpl.syncTableMetadata() Transactional(rollbackFor Exception.class) public synchronized Boolean syncTableMetadata(MetaSchema schema, String tableName, String dsName) { // 1. 同步表基本信息 ListMetaTable tables getTablesDynamic(catalog, schema, tableName, dsName); // 删除不存在的表 deleteTableNotExist(tableIdList); // upsert 表信息 saveOrUpdateBatch(tables); for (MetaTable table : tables) { // 2. 同步字段 ListMetaColumn columns metaColumnService.getColumnsDynamic(...); // 删除不存在的字段 → upsert 字段 // 3. 同步索引 ListMetaIndex indexes metaIndexService.getIndexInfo(...); // 4. 同步主键 ListMetaPrimaryKey pks metaPrimaryKeyService.getPrimaryKeys(...); // 5. 同步外键 ListMetaForeignKey fks metaForeignKeyService.getImportedKeys(...); } return true; }6. DatabaseMetaData 核心 API 对照表JDBC API返回用途getSchemas(catalog, schema)ResultSet获取数据库 Schema 列表getCatalogs()ResultSet获取 Catalog 列表MySQL 回退getTables(catalog, schema, table, types)ResultSet获取表/视图列表10 列getColumns(catalog, schema, table, column)ResultSet获取字段详情23 列getIndexInfo(catalog, schema, table, unique, approximate)ResultSet获取索引信息getPrimaryKeys(catalog, schema, table)ResultSet获取主键列getImportedKeys(catalog, schema, table)ResultSet获取外键列getDatabaseProductName()String获取数据库产品名getDatabaseProductVersion()String获取数据库版本7. 方案优势对比维度系统表方案JDBC DatabaseMetaData 方案兼容性每种数据库需单独编写 SQL一次编写所有 JDBC 驱动通用维护成本随数据库种类线性增长恒定类型映射需自行维护类型映射表getColumns()返回标准java.sql.Types新数据库接入需研究其系统表结构只需配置 JDBC 连接标准合规依赖数据库厂商实现遵循 JDBC 规范版本升级系统表结构可能变化JDBC 驱动保证向后兼容8. 注意事项MySQL 特殊处理MySQL 的getSchemas()可能返回空需回退到getCatalogs()空值安全ResultSet.getXXX()在值为NULL时会抛异常应使用JdbcUtils.getXXXMetaData()安全读取性能考量大表量场景建议按 Schema 分批同步避免一次性加载过多元数据连接管理务必使用 try-with-resources 确保 Connection / ResultSet 正确关闭租户隔离异步同步时通过TenantBroker.runAs()保证租户上下文在线程池中正确传递幂等设计通过Idempotent注解防止短时间内重复提交同步任务
