add metaData.addUpdateTableMateList(String tableName);

xlorne · xlorne · commit 3b3ba65124d4 · 2025-11-10T20:18:48.000+08:00
diff --git a/README.md b/README.md
@@ -174,13 +174,16 @@ List<DBMetaData> metaDataList = DBStreamContext.getInstance().metaDataList();
 // 通过 jdbcKey 获取指定数据库的元数据信息
 DBMetaData metaData = DBStreamContext.getInstance().getMetaData(String jdbcKey);
 
+// 主动更新表的元数据信息，在使用改变时将会先更新metadata然后再执行业务。（动态更新表的元数据）
+metaData.addUpdateTableMateList(String tableName);
+
 // 获取所有已缓存的数据库连接 jdbcKey 列表
 List<String> dbKeys = DBStreamContext.getInstance().loadDbKeys();
 
-// 清理指定数据库的元数据缓存（清空后下次访问会自动重新加载）
+// 清理指定数据库的元数据缓存（清空后下次访问会自动重新加载，动态更新表的元数据）
 DBStreamContext.getInstance().clear(String jdbcKey);
 
-// 清理所有数据库的元数据缓存
+// 清理所有数据库的元数据缓存（动态更新表的元数据）
 DBStreamContext.getInstance().clearAll();
 ```
 
diff --git a/pom.xml b/pom.xml
@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.codingapi.dbstream</groupId>
     <artifactId>dbstream-driver</artifactId>
-    <version>1.0.2</version>
+    <version>1.0.3</version>
 
     <url>https://github.com/codingapi/dbstream-driver</url>
     <name>dbstream-driver</name>
diff --git a/src/main/java/com/codingapi/dbstream/DBStreamContext.java b/src/main/java/com/codingapi/dbstream/DBStreamContext.java
@@ -42,7 +42,8 @@ public void addListener(SQLExecuteListener sqlExecuteListener) {
 
     /**
      * 是否支持SQL 拦截代理分析
-     * @param info 数据库连接信息
+     *
+     * @param info    数据库连接信息
      * @param dbTable 数据库表信息
      * @return 是否支持
      */
diff --git a/src/main/java/com/codingapi/dbstream/driver/DBStreamProxyDriver.java b/src/main/java/com/codingapi/dbstream/driver/DBStreamProxyDriver.java
@@ -27,15 +27,15 @@ public class DBStreamProxyDriver implements Driver {
 
     @Override
     public Connection connect(String url, Properties info) throws SQLException {
-        if(this.driver==null){
+        if (this.driver == null) {
             this.acceptsURL(url);
         }
         Connection connection = driver.connect(url, info);
         info.setProperty(DBMetaData.KEY_JDBC_URL, url);
         String jdbcKey = info.getProperty(DBMetaData.KEY_JDBC_KEY);
         DBMetaData metaData = DBMetaContext.getInstance().getMetaData(jdbcKey);
         if (metaData == null) {
-            DBScanner scanner = new DBScanner(connection,info);
+            DBScanner scanner = new DBScanner(connection, info);
             metaData = scanner.loadMetadata();
             DBMetaContext.getInstance().update(metaData);
         }
diff --git a/src/main/java/com/codingapi/dbstream/interceptor/SQLExecuteState.java b/src/main/java/com/codingapi/dbstream/interceptor/SQLExecuteState.java
@@ -2,6 +2,7 @@
 
 import com.codingapi.dbstream.proxy.ConnectionProxy;
 import com.codingapi.dbstream.scanner.DBMetaData;
+import com.codingapi.dbstream.scanner.DBScanner;
 import com.codingapi.dbstream.scanner.DbColumn;
 import com.codingapi.dbstream.scanner.DbTable;
 import lombok.Getter;
@@ -127,19 +128,6 @@ public void setParam(int index, Object value) {
     }
 
 
-    /**
-     * 获取表的元数据数据信息
-     *
-     * @param tableName 表名称 不区分大小写
-     * @return 表数据
-     */
-    public DbTable getTable(String tableName) {
-        if (metaData == null) {
-            return null;
-        }
-        return metaData.getTable(tableName);
-    }
-
     /**
      * 元数据中是否存在表
      *
@@ -172,7 +160,7 @@ public List<Map<String, Object>> query(String sql, List<Object> params) throws S
         PreparedStatement preparedStatement = connection.prepareStatement(sql);
         for (int i = 0; i < params.size(); i++) {
             Object param = params.get(i);
-            preparedStatement.setObject(i + 1,param);
+            preparedStatement.setObject(i + 1, param);
         }
         ResultSet resultSet = preparedStatement.executeQuery();
         ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
@@ -216,18 +204,39 @@ public List<Map<String, Object>> getStatementGenerateKeys(DbTable dbTable) {
     }
 
 
-    public Properties getDriverProperties(){
+    /**
+     * 获取驱动配置信息
+     * @return Properties
+     */
+    public Properties getDriverProperties() {
         if (metaData == null) {
             return null;
         }
         return this.metaData.getProperties();
     }
 
 
+    /**
+     * 获取数据库的jdbcUrl
+     * @return jdbcUrl
+     */
     public String getJdbcUrl() {
         if (metaData == null) {
             return null;
         }
         return metaData.getJdbcUrl();
     }
+
+
+    /**
+     * 更新数据库的元数据信息
+     * @param tableName 表名
+     */
+    public void updateMetaData(String tableName) throws SQLException {
+        // 当前表需要更新时，将会连同所有带更新的表一次性全部更新
+        if (this.metaData.isUpdateTableMeta(tableName)) {
+            DBScanner dbScanner = new DBScanner(connection, getDriverProperties());
+            dbScanner.updateMetadata(this.metaData);
+        }
+    }
 }
diff --git a/src/main/java/com/codingapi/dbstream/listener/SQLDeleteExecuteListener.java b/src/main/java/com/codingapi/dbstream/listener/SQLDeleteExecuteListener.java
@@ -28,6 +28,7 @@ public void before(SQLExecuteState executeState) throws SQLException {
                 Statement parserStatement = CCJSqlParserUtil.parse(sql);
                 Delete delete = (Delete) parserStatement;
                 Table table = delete.getTable();
+                executeState.updateMetaData(table.getName());
                 DbTable dbTable = executeState.getDbTable(table.getName());
                 if (dbTable != null && DBStreamContext.getInstance().support(executeState.getDriverProperties(), dbTable)) {
                     DeleteDBEventParser dataParser = new DeleteDBEventParser(executeState, delete, table, dbTable);
diff --git a/src/main/java/com/codingapi/dbstream/listener/SQLInsertExecuteListener.java b/src/main/java/com/codingapi/dbstream/listener/SQLInsertExecuteListener.java
@@ -28,6 +28,7 @@ public void before(SQLExecuteState executeState) throws SQLException {
                 Statement parserStatement = CCJSqlParserUtil.parse(sql);
                 Insert insert = (Insert) parserStatement;
                 Table table = insert.getTable();
+                executeState.updateMetaData(table.getName());
                 DbTable dbTable = executeState.getDbTable(table.getName());
                 if (dbTable != null && DBStreamContext.getInstance().support(executeState.getDriverProperties(), dbTable)) {
                     InsertDBEventParser dataParser = new InsertDBEventParser(executeState, insert, table, dbTable);
diff --git a/src/main/java/com/codingapi/dbstream/listener/SQLUpdateExecuteListener.java b/src/main/java/com/codingapi/dbstream/listener/SQLUpdateExecuteListener.java
@@ -28,6 +28,7 @@ public void before(SQLExecuteState executeState) throws SQLException {
                 Statement parserStatement = CCJSqlParserUtil.parse(sql);
                 Update update = (Update) parserStatement;
                 Table table = update.getTable();
+                executeState.updateMetaData(table.getName());
                 DbTable dbTable = executeState.getDbTable(table.getName());
                 if (dbTable != null && DBStreamContext.getInstance().support(executeState.getDriverProperties(), dbTable)) {
                     UpdateDBEventParser dataParser = new UpdateDBEventParser(executeState, update, table, dbTable);
diff --git a/src/main/java/com/codingapi/dbstream/parser/InsertDBEventParser.java b/src/main/java/com/codingapi/dbstream/parser/InsertDBEventParser.java
@@ -115,13 +115,13 @@ private List<DBEvent> loadDefaultInsertEvent() {
 
 
     private List<String> loadInsertColumns() {
-        ExpressionList<?> values =  this.insert.getValues().getExpressions();
+        ExpressionList<?> values = this.insert.getValues().getExpressions();
         ExpressionList<Column> columns = this.insert.getColumns();
         List<String> columnList = new ArrayList<>();
-        for (int i=0;i<columns.size();i++) {
+        for (int i = 0; i < columns.size(); i++) {
             Column column = columns.get(i);
             Expression expression = values.get(i);
-            if(expression instanceof JdbcParameter) {
+            if (expression instanceof JdbcParameter) {
                 columnList.add(column.getColumnName());
             }
         }
diff --git a/src/main/java/com/codingapi/dbstream/provider/DefaultDBTableSupportProvider.java b/src/main/java/com/codingapi/dbstream/provider/DefaultDBTableSupportProvider.java
@@ -8,7 +8,7 @@ public class DefaultDBTableSupportProvider implements DBTableSupportProvider {
 
     @Override
     public boolean support(Properties info, DbTable dbTable) {
-        if(dbTable.hasColumns() && dbTable.hasPrimaryKeys()) {
+        if (dbTable.hasColumns() && dbTable.hasPrimaryKeys()) {
             return true;
         }
         return false;
diff --git a/src/main/java/com/codingapi/dbstream/scanner/DBMetaContext.java b/src/main/java/com/codingapi/dbstream/scanner/DBMetaContext.java
@@ -45,7 +45,7 @@ public void update(DBMetaData metaData) {
      */
     public void clearAll() {
         for (DBMetaData metaData : this.cache.values()) {
-            metaData.clean();
+            metaData.cleanSerializable();
         }
         cache.clear();
     }
@@ -80,7 +80,7 @@ public List<DBMetaData> metaDataList() {
     public void clear(String jdbcKey) {
         DBMetaData dbMetaData = cache.get(jdbcKey);
         if (dbMetaData != null) {
-            dbMetaData.clean();
+            dbMetaData.cleanSerializable();
         }
         cache.remove(jdbcKey);
     }
diff --git a/src/main/java/com/codingapi/dbstream/scanner/DBMetaData.java b/src/main/java/com/codingapi/dbstream/scanner/DBMetaData.java
@@ -4,15 +4,20 @@
 import com.codingapi.dbstream.utils.SHA256Util;
 import lombok.Getter;
 
-import java.util.List;
-import java.util.Properties;
+import java.util.*;
 
 
 public class DBMetaData {
 
     public static final String KEY_JDBC_URL = "jdbc.url";
     public static final String KEY_JDBC_KEY = "jdbc.key";
 
+    /**
+     * 待更新的table meta数据
+     */
+    @Getter
+    private final List<String> updateTableMetaList = new ArrayList<>();
+
     /**
      * 创建唯一标识
      *
@@ -24,12 +29,21 @@ private String generateKey(String schema) {
         return SHA256Util.sha256(data);
     }
 
+    /**
+     * 数据记录时间
+     */
     @Getter
     private long updateTime = 0;
 
+    /**
+     * 扫描的表元数据信息
+     */
     @Getter
     private List<DbTable> tables;
 
+    /**
+     * jdbc 的配置信息
+     */
     @Getter
     private final Properties properties;
 
@@ -38,6 +52,17 @@ public DBMetaData(Properties properties, String schema) {
         this.properties.setProperty(DBMetaData.KEY_JDBC_KEY, this.generateKey(schema));
     }
 
+    /**
+     * 更新表的元数据信息
+     *
+     * @param tableName 表名称
+     */
+    public void addUpdateTableMateList(String tableName) {
+        String upTableName = tableName.toUpperCase();
+        if (this.updateTableMetaList.contains(upTableName)) {
+            this.updateTableMetaList.add(upTableName);
+        }
+    }
 
     void setTables(List<DbTable> tables) {
         this.tables = tables;
@@ -51,6 +76,12 @@ public boolean isEmpty() {
         return this.tables == null || this.tables.isEmpty();
     }
 
+    /**
+     * 获取表的元信息
+     *
+     * @param tableName 表名称
+     * @return DbTable
+     */
     public DbTable getTable(String tableName) {
         if (this.tables == null) {
             return null;
@@ -64,24 +95,74 @@ public DbTable getTable(String tableName) {
         return null;
     }
 
-    public void clean(){
+    /**
+     * 清空序列化的本地缓存数据
+     */
+    void cleanSerializable() {
         DBTableSerializableHelper tableSerializableHelper = new DBTableSerializableHelper(this.getKeyJdbcKey());
         tableSerializableHelper.clean();
         this.tables = null;
     }
 
+    /**
+     * 获取jdbc的url
+     *
+     * @return jdbcUrl
+     */
     public String getJdbcUrl() {
         if (properties == null) {
             return null;
         }
         return properties.getProperty(DBMetaData.KEY_JDBC_URL);
     }
 
+    /**
+     * 获取jdbc的唯一值
+     *
+     * @return jdbcKey
+     */
     public String getKeyJdbcKey() {
         if (properties == null) {
             return null;
         }
         return properties.getProperty(DBMetaData.KEY_JDBC_KEY);
     }
 
+    /**
+     * 是否为待更新的数据库表
+     *
+     * @param tableName 表名称
+     * @return true 是
+     */
+    public boolean isUpdateTableMeta(String tableName) {
+        return this.updateTableMetaList.contains(tableName.toUpperCase());
+    }
+
+    /**
+     * 更新DbTable数据
+     *
+     * @param updateList 更新的表元数据信息
+     */
+    void updateDbTable(List<DbTable> updateList) {
+        List<DbTable> list = new ArrayList<>();
+        Map<String, DbTable> updateDbTables = new HashMap<>();
+        for (DbTable update : updateList) {
+            String tableName = update.getName().toUpperCase();
+            updateDbTables.put(tableName, update);
+            this.updateTableMetaList.remove(tableName);
+        }
+
+        for (DbTable dbTable : tables) {
+            String tableName = dbTable.getName().toUpperCase();
+            DbTable updateTable = updateDbTables.get(tableName);
+            if (updateTable != null) {
+                list.add(updateTable);
+            } else {
+                list.add(dbTable);
+            }
+        }
+
+        this.tables.clear();
+        this.tables = list;
+    }
 }
diff --git a/src/main/java/com/codingapi/dbstream/scanner/DBScanner.java b/src/main/java/com/codingapi/dbstream/scanner/DBScanner.java
@@ -8,6 +8,7 @@
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Properties;
 
 public class DBScanner {
@@ -91,4 +92,31 @@ public DBMetaData loadMetadata() throws SQLException {
     }
 
 
+    /**
+     * 更新对应表的元数据信息
+     *
+     * @param dbMetaData 数据库下的所有元数据信息
+     * @throws SQLException SQLException
+     */
+    public void updateMetadata(DBMetaData dbMetaData) throws SQLException {
+        DatabaseMetaData metaData = connection.getMetaData();
+        String catalog = connection.getCatalog();
+        String schema = connection.getSchema();
+        ResultSet tables = metaData.getTables(catalog, schema, "%", new String[]{"TABLE"});
+        List<DbTable> updateList = new ArrayList<>();
+        while (tables.next()) {
+            String tableName = tables.getString("TABLE_NAME");
+            String remarks = tables.getString("REMARKS");
+            if (dbMetaData.isUpdateTableMeta(tableName)) {
+                DbTable tableInfo = new DbTable(tableName, remarks);
+                this.loadDbTableInfo(tableName, tableInfo);
+                updateList.add(tableInfo);
+            }
+        }
+        tables.close();
+        dbMetaData.updateDbTable(updateList);
+        dbMetaData.success();
+    }
+
+
 }
diff --git a/src/main/java/com/codingapi/dbstream/scanner/DbTable.java b/src/main/java/com/codingapi/dbstream/scanner/DbTable.java
diff --git a/src/main/java/com/codingapi/dbstream/stream/DBEventContext.java b/src/main/java/com/codingapi/dbstream/stream/DBEventContext.java
diff --git a/src/main/java/com/codingapi/dbstream/stream/DefaultDBEventPusher.java b/src/main/java/com/codingapi/dbstream/stream/DefaultDBEventPusher.java

Original file line number	Diff line number	Diff line change
`@@ -8,7 +8,7 @@ public class DefaultDBTableSupportProvider implements DBTableSupportProvider {`
`8`	`8`
`9`	`9`	`@Override`
`10`	`10`	`public boolean support(Properties info, DbTable dbTable) {`
`11`		`- if(dbTable.hasColumns() && dbTable.hasPrimaryKeys()) {`
	`11`	`+ if (dbTable.hasColumns() && dbTable.hasPrimaryKeys()) {`
`12`	`12`	`return true;`
`13`	`13`	`}`
`14`	`14`	`return false;`