更新数据同步
This commit is contained in:
215
src/main/java/com/mini/capi/utils/MysqlUtils.java
Normal file
215
src/main/java/com/mini/capi/utils/MysqlUtils.java
Normal file
@@ -0,0 +1,215 @@
|
||||
package com.mini.capi.utils;
|
||||
|
||||
import com.mini.capi.biz.domain.BizDbConfig;
|
||||
import com.mini.capi.biz.domain.DataTableField;
|
||||
import com.mini.capi.biz.domain.DataTableInfo;
|
||||
import com.mini.capi.model.info.TableTree;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.math.RoundingMode;
|
||||
import java.sql.*;
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.ZoneId;
|
||||
import java.util.*;
|
||||
import java.util.Date;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
public class MysqlUtils {
|
||||
|
||||
private static final LoggerUtils logger = LoggerUtils.getInstance();
|
||||
|
||||
// 需要排除的系统数据库
|
||||
private static final List<String> SYSTEM_DATABASES = Arrays.asList(
|
||||
"information_schema", "mysql", "performance_schema", "sys", "test"
|
||||
);
|
||||
|
||||
// 提取字段长度的正则表达式（如varchar(50) -> 50）
|
||||
private static final Pattern LENGTH_PATTERN = Pattern.compile("\\((\\d+)\\)");
|
||||
|
||||
/**
|
||||
* 封装：获取MySQL数据库连接
|
||||
*/
|
||||
private static Connection getConnection(String ip, Integer port, String username, String password) throws Exception {
|
||||
String driver = "com.mysql.cj.jdbc.Driver";
|
||||
String jdbcUrl = String.format(
|
||||
"jdbc:mysql://%s:%d/information_schema?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true",
|
||||
ip, port
|
||||
);
|
||||
Class.forName(driver); // 加载驱动
|
||||
return DriverManager.getConnection(jdbcUrl, username, password);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取指定MySQL连接的所有非系统库表结构信息
|
||||
*
|
||||
* @param ip MySQL服务器IP
|
||||
* @param port 端口（默认3306）
|
||||
* @param username 用户名
|
||||
* @param password 密码
|
||||
* @return 数据库名 -> 表信息列表（包含字段）的映射
|
||||
* @throws Exception 连接或查询异常
|
||||
*/
|
||||
public static Map<String, List<DataTableInfo>> getMysqlSchemaInfo(Connection conn) throws Exception {
|
||||
Map<String, List<DataTableInfo>> result = new HashMap<>();
|
||||
// 1. 获取所有非系统数据库
|
||||
List<String> databases = getNonSystemDatabases(conn);
|
||||
logger.info("获取到非系统数据库数量：", databases.size());
|
||||
// 2. 遍历数据库，获取表和字段信息
|
||||
for (String dbName : databases) {
|
||||
List<DataTableInfo> tableInfos = getTablesByDatabase(conn, dbName);
|
||||
result.put(dbName, tableInfos);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取所有非系统数据库
|
||||
*/
|
||||
private static List<String> getNonSystemDatabases(Connection conn) throws SQLException {
|
||||
List<String> databases = new ArrayList<>();
|
||||
String sql = "SELECT SCHEMA_NAME FROM SCHEMATA WHERE SCHEMA_NAME NOT IN ("
|
||||
+ String.join(",", Collections.nCopies(SYSTEM_DATABASES.size(), "?")) + ")";
|
||||
try (PreparedStatement ps = conn.prepareStatement(sql)) {
|
||||
for (int i = 0; i < SYSTEM_DATABASES.size(); i++) {
|
||||
ps.setString(i + 1, SYSTEM_DATABASES.get(i));
|
||||
}
|
||||
try (ResultSet rs = ps.executeQuery()) {
|
||||
while (rs.next()) {
|
||||
databases.add(rs.getString("SCHEMA_NAME"));
|
||||
}
|
||||
}
|
||||
}
|
||||
return databases;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取指定数据库下的所有表信息（包含字段）
|
||||
*/
|
||||
private static List<DataTableInfo> getTablesByDatabase(Connection conn, String dbName) throws SQLException {
|
||||
List<DataTableInfo> tableInfos = new ArrayList<>();
|
||||
String tableSql = "SELECT " +
|
||||
"TABLE_NAME, TABLE_COMMENT, CREATE_TIME, UPDATE_TIME, " +
|
||||
"DATA_LENGTH, INDEX_LENGTH, TABLE_ROWS " +
|
||||
"FROM TABLES WHERE TABLE_SCHEMA = ?";
|
||||
try (PreparedStatement tablePs = conn.prepareStatement(tableSql)) {
|
||||
tablePs.setString(1, dbName);
|
||||
try (ResultSet tableRs = tablePs.executeQuery()) {
|
||||
while (tableRs.next()) {
|
||||
DataTableInfo tableInfo = buildDataTableInfo(tableRs, dbName);
|
||||
List<DataTableField> fields = getFieldsByTable(conn, dbName, tableInfo.getTableName());
|
||||
fields.forEach(field -> field.setTableId(tableInfo.getTableId()));
|
||||
tableInfos.add(tableInfo);
|
||||
}
|
||||
}
|
||||
}
|
||||
return tableInfos;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建DataTableInfo实体
|
||||
*/
|
||||
private static DataTableInfo buildDataTableInfo(ResultSet tableRs, String dbName) throws SQLException {
|
||||
DataTableInfo tableInfo = new DataTableInfo();
|
||||
tableInfo.setTableId(vId.getUid());
|
||||
tableInfo.setTableName(tableRs.getString("TABLE_NAME"));
|
||||
tableInfo.setTableComment(tableRs.getString("TABLE_COMMENT"));
|
||||
long dataLength = tableRs.getLong("DATA_LENGTH");
|
||||
long indexLength = tableRs.getLong("INDEX_LENGTH");
|
||||
BigDecimal tableSize = BigDecimal.valueOf((dataLength + indexLength) / 1024.0 / 1024.0)
|
||||
.setScale(2, RoundingMode.HALF_UP);
|
||||
tableInfo.setTableSize(tableSize);
|
||||
tableInfo.setDataSource(dbName);
|
||||
tableInfo.setDataRows(tableRs.getLong("TABLE_ROWS"));
|
||||
Date createDate = tableRs.getTimestamp("CREATE_TIME");
|
||||
if (createDate != null) {
|
||||
tableInfo.setCreateTime(createDate.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime());
|
||||
}
|
||||
Date updateDate = tableRs.getTimestamp("UPDATE_TIME");
|
||||
if (updateDate != null) {
|
||||
tableInfo.setUpdateTime(updateDate.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime());
|
||||
}
|
||||
tableInfo.setDs(DateUtils.dsValue());
|
||||
return tableInfo;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取指定表的字段信息
|
||||
*/
|
||||
private static List<DataTableField> getFieldsByTable(Connection conn, String dbName, String tableName) throws SQLException {
|
||||
List<DataTableField> fields = new ArrayList<>();
|
||||
String fieldSql = "SELECT " +
|
||||
"TABLE_SCHEMA,TABLE_NAME,COLUMN_NAME, COLUMN_TYPE, COLUMN_COMMENT, " +
|
||||
"ORDINAL_POSITION, CHARACTER_MAXIMUM_LENGTH " +
|
||||
"FROM COLUMNS WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? " +
|
||||
"ORDER BY ORDINAL_POSITION";
|
||||
try (PreparedStatement fieldPs = conn.prepareStatement(fieldSql)) {
|
||||
fieldPs.setString(1, dbName);
|
||||
fieldPs.setString(2, tableName);
|
||||
try (ResultSet fieldRs = fieldPs.executeQuery()) {
|
||||
while (fieldRs.next()) {
|
||||
fields.add(buildDataTableField(fieldRs));
|
||||
}
|
||||
}
|
||||
}
|
||||
return fields;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建DataTableField实体
|
||||
*/
|
||||
private static DataTableField buildDataTableField(ResultSet fieldRs) throws SQLException {
|
||||
DataTableField field = new DataTableField();
|
||||
field.setFieldId(vId.getUid());
|
||||
field.setDataSource(fieldRs.getString("TABLE_SCHEMA"));
|
||||
field.setTableName(fieldRs.getString("TABLE_NAME"));
|
||||
field.setFieldName(fieldRs.getString("COLUMN_NAME"));
|
||||
String fieldType = fieldRs.getString("COLUMN_TYPE");
|
||||
field.setFieldType(fieldType);
|
||||
field.setFieldOrder(fieldRs.getInt("ORDINAL_POSITION"));
|
||||
field.setFieldRemark(fieldRs.getString("COLUMN_COMMENT"));
|
||||
Long length = fieldRs.getLong("CHARACTER_MAXIMUM_LENGTH");
|
||||
if (length == 0 || fieldRs.wasNull()) {
|
||||
length = extractLengthFromType(fieldType);
|
||||
}
|
||||
field.setFieldLength(length);
|
||||
field.setCreateTime(LocalDateTime.now());
|
||||
field.setDs(DateUtils.dsValue());
|
||||
return field;
|
||||
}
|
||||
|
||||
/**
|
||||
* 从字段类型中提取长度（如int(11) -> 11）
|
||||
*/
|
||||
private static Long extractLengthFromType(String fieldType) {
|
||||
if (fieldType == null) return null;
|
||||
Matcher matcher = LENGTH_PATTERN.matcher(fieldType);
|
||||
if (matcher.find()) {
|
||||
try {
|
||||
return Long.parseLong(matcher.group(1));
|
||||
} catch (NumberFormatException e) {
|
||||
logger.warn("提取字段长度失败，类型", fieldType, e);
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public static List<TableTree> getTableTrees(BizDbConfig dbConfig) {
|
||||
List<TableTree> tableTrees = new ArrayList<>();
|
||||
try {
|
||||
Connection conn = getConnection(dbConfig.getDbIp(), dbConfig.getDbPort(), dbConfig.getDbUsername(), dbConfig.getDbPassword());
|
||||
Map<String, List<DataTableInfo>> schemaInfo = MysqlUtils.getMysqlSchemaInfo(conn);
|
||||
for (Map.Entry<String, List<DataTableInfo>> entry : schemaInfo.entrySet()) {
|
||||
for (DataTableInfo tableInfo : entry.getValue()) {
|
||||
tableInfo.setDbId(dbConfig.getId());
|
||||
List<DataTableField> dataTableFields = getFieldsByTable(conn, entry.getKey(), tableInfo.getTableName());
|
||||
dataTableFields.stream().forEach(tableField -> tableField.setTableId(tableInfo.getTableId()));
|
||||
tableTrees.add(new TableTree(tableInfo, dataTableFields));
|
||||
}
|
||||
}
|
||||
} catch (Exception e) {
|
||||
logger.error(e.getMessage());
|
||||
}
|
||||
return tableTrees;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user