From 9fea8d19cdf137dd87cf6af5520e8e252268c1f9 Mon Sep 17 00:00:00 2001 From: lorne <1991wangliang@gmail.com> Date: Mon, 29 Jun 2026 11:25:34 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E6=89=A7=E8=A1=8C=E7=BB=93=E6=9E=9C=E6=95=B0=E7=BB=84=E4=B8=8E?= =?UTF-8?q?=E6=89=B9=E9=87=8F=E8=A1=8C=E6=95=B0=E4=B8=8D=E4=B8=80=E8=87=B4?= =?UTF-8?q?=E6=97=B6=E7=9A=84=E8=B6=8A=E7=95=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit executeBatch()/executeLargeBatch() 在 MySQL rewriteBatchedStatements 等 场景下返回的结果数组长度可能小于批量行数，导致 DBEventListener.after() 中 arrays.get(i) 抛 IndexOutOfBoundsException。 - batchResultToArrays 现恒定返回长度等于批量行数的 List，缺失位置按 「成功但行数未知」补齐，避免越界且事件不丢失。 - ResultSetUtils.isNotUpdatedRows 将 SUCCESS_NO_INFO(-2) 视为已更新， 避免整批事件被静默丢弃。 - 新增 DBEventListenerBatchResultTest 覆盖越界回归。 Co-Authored-By: Claude --- .../listener/dbevent/DBEventListener.java | 17 ++-- .../dbstream/utils/ResultSetUtils.java | 14 ++- .../DBEventListenerBatchResultTest.java | 92 +++++++++++++++++++ 3 files changed, 116 insertions(+), 7 deletions(-) create mode 100644 src/test/java/com/codingapi/dbstream/listener/dbevent/DBEventListenerBatchResultTest.java diff --git a/src/main/java/com/codingapi/dbstream/listener/dbevent/DBEventListener.java b/src/main/java/com/codingapi/dbstream/listener/dbevent/DBEventListener.java index c5ca29f..55ae601 100644 --- a/src/main/java/com/codingapi/dbstream/listener/dbevent/DBEventListener.java +++ b/src/main/java/com/codingapi/dbstream/listener/dbevent/DBEventListener.java @@ -81,21 +81,26 @@ public void before(SQLRunningState runningState) throws SQLException { /** - * batch result 结果数组转化 + * batch result 结果数组转化。 + *

+ * 返回的 List 长度恒等于批量行数 {@code size}。当 JDBC 驱动返回的结果数组长度 + * 与批量行数不一致时（如 MySQL {@code rewriteBatchedStatements=true} 会将批量语句 + * 合并/拆分执行，返回数组长度小于批量行数），对缺失位置按「成功但行数未知」补齐(1)， + * 既避免 {@link List#get(int)} 越界，也保证对应行的事件不被丢弃。 */ - private List batchResultToArrays(Object result, int size) { + List batchResultToArrays(Object result, int size) { List list = new ArrayList<>(); if (result instanceof int[]) { int[] rows = (int[]) result; - for (int row : rows) { - list.add(row); + for (int i = 0; i < size; i++) { + list.add(i < rows.length ? rows[i] : 1); } return list; } if (result instanceof long[]) { long[] rows = (long[]) result; - for (long row : rows) { - list.add(row); + for (int i = 0; i < size; i++) { + list.add(i < rows.length ? rows[i] : 1L); } return list; } diff --git a/src/main/java/com/codingapi/dbstream/utils/ResultSetUtils.java b/src/main/java/com/codingapi/dbstream/utils/ResultSetUtils.java index c9fc076..1750cf0 100644 --- a/src/main/java/com/codingapi/dbstream/utils/ResultSetUtils.java +++ b/src/main/java/com/codingapi/dbstream/utils/ResultSetUtils.java @@ -1,21 +1,33 @@ package com.codingapi.dbstream.utils; +import java.sql.Statement; + /** * 执行结果判断 */ public class ResultSetUtils { /** - * 是否执行不成功，受影响数据是否小于等于0 + * 是否执行不成功，受影响数据是否小于等于0。 + *

+ * 批量执行(executeBatch/executeLargeBatch)中，JDBC 驱动常用 + * {@link Statement#SUCCESS_NO_INFO}(-2) 表示「执行成功但具体行数未知」， + * 此时应视为已更新，否则会导致整批事件被静默丢弃。 */ public static boolean isNotUpdatedRows(Object result) { if (result instanceof Integer) { int value = (Integer) result; + if (value == Statement.SUCCESS_NO_INFO) { + return false; + } return value <= 0; } if (result instanceof Long) { long value = (Long) result; + if (value == Statement.SUCCESS_NO_INFO) { + return false; + } return value <= 0; } return false; diff --git a/src/test/java/com/codingapi/dbstream/listener/dbevent/DBEventListenerBatchResultTest.java b/src/test/java/com/codingapi/dbstream/listener/dbevent/DBEventListenerBatchResultTest.java new file mode 100644 index 0000000..cdccbf0 --- /dev/null +++ b/src/test/java/com/codingapi/dbstream/listener/dbevent/DBEventListenerBatchResultTest.java @@ -0,0 +1,92 @@ +package com.codingapi.dbstream.listener.dbevent; + +import com.codingapi.dbstream.listener.SQLRunningState; +import com.codingapi.dbstream.parser.DBEventParser; +import com.codingapi.dbstream.parser.SQLParser; +import com.codingapi.dbstream.scanner.DbTable; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * {@link DBEventListener#batchResultToArrays(Object, int)} 规范化逻辑测试。 + * 重点回归：JDBC 驱动返回的结果数组长度与批量行数不一致时，不得越界且事件不得丢失。 + */ +class DBEventListenerBatchResultTest { + + private DBEventListener newListener() { + return new DBEventListener() { + @Override + public int order() { + return 0; + } + + @Override + public boolean support(String sql) { + return false; + } + + @Override + public SQLParser createSQLParser(String sql) { + return null; + } + + @Override + public DBEventParser createDbEventParser(SQLRunningState runningState, SQLParser sqlParser, DbTable dbTable) { + return null; + } + }; + } + + @Test + void intArrayLengthEqualsSize() { + DBEventListener listener = newListener(); + List arrays = listener.batchResultToArrays(new int[]{1, 1, 1}, 3); + assertEquals(3, arrays.size()); + assertEquals(1, arrays.get(0)); + assertEquals(1, arrays.get(2)); + } + + /** + * 回归 rewriteBatchedStatements 等场景：驱动返回数组长度小于批量行数， + * 修复前 List 长度=驱动数组长度，after() 中 arrays.get(i) 会 IndexOutOfBounds。 + */ + @Test + void intArrayLengthLessThanSize_notOutOfBounds() { + DBEventListener listener = newListener(); + // 批量行数 636，驱动合并后仅返回 635 个结果（复现线上越界场景） + List arrays = listener.batchResultToArrays(new int[]{1, 1, 1}, 636); + assertEquals(636, arrays.size()); + // 末位由「成功」哨兵补齐，必须可安全访问 + assertEquals(1, arrays.get(635)); + } + + @Test + void intArrayLengthGreaterThanSize_truncated() { + DBEventListener listener = newListener(); + List arrays = listener.batchResultToArrays(new int[]{1, 2, 3, 4}, 2); + assertEquals(2, arrays.size()); + assertEquals(1, arrays.get(0)); + assertEquals(2, arrays.get(1)); + } + + @Test + void longArrayLengthLessThanSize_notOutOfBounds() { + DBEventListener listener = newListener(); + List arrays = listener.batchResultToArrays(new long[]{1L, 2L}, 4); + assertEquals(4, arrays.size()); + assertEquals(1L, arrays.get(0)); + assertEquals(1L, arrays.get(3)); + } + + @Test + void nonArrayResult_fallbackAllZero() { + DBEventListener listener = newListener(); + List arrays = listener.batchResultToArrays(5, 3); + assertEquals(3, arrays.size()); + assertEquals(0, arrays.get(0)); + assertEquals(0, arrays.get(2)); + } +} From ad14e5c2c6b33af99513938c1876dbb780c45763 Mon Sep 17 00:00:00 2001 From: lorne <1991wangliang@gmail.com> Date: Mon, 29 Jun 2026 11:28:22 +0800 Subject: [PATCH 2/2] =?UTF-8?q?chore:=20=E5=8D=87=E7=BA=A7=E7=89=88?= =?UTF-8?q?=E6=9C=AC=E8=87=B3=201.0.19=20=E5=B9=B6=E6=B8=85=E7=90=86?= =?UTF-8?q?=E6=97=A7=E6=96=87=E6=A1=A3=E4=B8=8E=E6=9C=AC=E5=9C=B0=E9=85=8D?= =?UTF-8?q?=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pom.xml 版本号 1.0.18 → 1.0.19 - 移除 docs/ 下旧的架构/组件/变更/计划文档及 .claude/settings.json Co-Authored-By: Claude --- .claude/settings.json | 6 - docs/architecture/dbstream-driver.md | 179 ----------------- .../819340-dynamic-table-metadata-refresh.md | 42 ---- .../dbstream-driver_DBEventPusher.md | 187 ------------------ .../dbstream-driver_DBEventSupporter.md | 161 --------------- .../dbstream-driver_DBStreamContext.md | 183 ----------------- .../dbstream-driver_SQLExecuteListener.md | 168 ---------------- docs/components/index.md | 11 -- ...0-change-dynamic-table-metadata-refresh.md | 96 --------- pom.xml | 2 +- 10 files changed, 1 insertion(+), 1034 deletions(-) delete mode 100644 .claude/settings.json delete mode 100644 docs/architecture/dbstream-driver.md delete mode 100644 docs/changes/819340-dynamic-table-metadata-refresh.md delete mode 100644 docs/components/dbstream-driver_DBEventPusher.md delete mode 100644 docs/components/dbstream-driver_DBEventSupporter.md delete mode 100644 docs/components/dbstream-driver_DBStreamContext.md delete mode 100644 docs/components/dbstream-driver_SQLExecuteListener.md delete mode 100644 docs/components/index.md delete mode 100644 docs/plan/819340-change-dynamic-table-metadata-refresh.md diff --git a/.claude/settings.json b/.claude/settings.json deleted file mode 100644 index a55708b..0000000 --- a/.claude/settings.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "enabledPlugins": { - "commons@coding-marketplace": true, - "backend@coding-marketplace": true - } -} diff --git a/docs/architecture/dbstream-driver.md b/docs/architecture/dbstream-driver.md deleted file mode 100644 index 3087401..0000000 --- a/docs/architecture/dbstream-driver.md +++ /dev/null @@ -1,179 +0,0 @@ -# dbstream-driver - -基于 JDBC 代理驱动的 Change Data Capture（CDC）组件，通过拦截 SQL 操作捕获数据库变更并发布结构化事件。零运行时依赖。 - -## Maven 坐标 - -- groupId: com.codingapi.dbstream -- artifactId: dbstream-driver -- version: 1.0.18 -- packaging: jar - -## 关联关系 - -> 以下关系由 `mvn dependency:tree` 指令结果生成，非人工推断。 - -### 我被哪些模块依赖 - -本项目为单模块工程，无 reactor 内部依赖方。 - -### 我依赖哪些模块 - -无 reactor 内部依赖。 - -### 主要外部依赖 - -- `org.projectlombok:lombok`（optional，编译期注解处理） -- 无其他运行期依赖 - -## 项目结构 - -``` -com.codingapi.dbstream/ -├── DBStreamContext.java # 对外统一入口（单例） -├── driver/ -│ └── DBStreamProxyDriver.java # java.sql.Driver 实现，SPI 注册 -├── proxy/ -│ ├── ConnectionProxy.java # 代理 Connection，拦截 commit/rollback/close -│ ├── PreparedStatementProxy.java # 代理 PreparedStatement，捕获参数与执行 -│ ├── StatementProxy.java # 代理 Statement -│ └── CallableStatementProxy.java # 代理 CallableStatement -├── parser/ -│ ├── SQLParser.java # SQL 解析接口（提取表名） -│ ├── InsertSQLParser.java # INSERT SQL 正则解析 -│ ├── UpdateSQLParser.java # UPDATE SQL 正则解析 -│ ├── DeleteSQLParser.java # DELETE SQL 正则解析 -│ ├── DBEventParser.java # 事件解析接口（prepare + loadEvents） -│ ├── InsertDBEventParser.java # INSERT 事件解析 -│ ├── UpdateDBEventParser.java # UPDATE 事件解析（查 pre-image/post-image） -│ └── DeleteDBEventParser.java # DELETE 事件解析（查 pre-image） -├── event/ -│ ├── EventType.java # 枚举：INSERT / UPDATE / DELETE -│ ├── DBEvent.java # 数据变更事件（表名、类型、数据、主键、事务标识） -│ ├── DBEventPusher.java # 事件推送接口 -│ ├── DefaultDBEventPusher.java # 默认推送实现（stdout） -│ ├── DBEventContext.java # 推送者注册与事件分发 -│ └── TransactionEventPools.java # ThreadLocal 事务级事件池 -├── listener/ -│ ├── SQLExecuteListener.java # SQL 执行监听接口（order/before/after） -│ ├── SQLRunningContext.java # 监听器注册中心，编排 before/after 回调 -│ ├── SQLRunningState.java # 单次执行上下文（SQL、参数、元数据） -│ ├── SQLRunningParam.java # 参数追踪（index/key 两种模式） -│ └── dbevent/ -│ ├── DBEventListener.java # 抽象基类，串联 SQL 解析→事件生成→事务池 -│ ├── InsertEventListener.java # INSERT 监听器（order=100） -│ ├── UpdateEventListener.java # UPDATE 监听器（order=100） -│ ├── DeleteEventListener.java # DELETE 监听器（order=100） -│ └── DBEventCacheContext.java # ThreadLocal 缓存 DBEventParser（支持批量） -├── scanner/ -│ ├── DBScanner.java # 数据库元数据扫描（DatabaseMetaData），支持全量/单表/按名称列表扫描 -│ ├── DBMetaData.java # 单数据源全部表元数据，支持按表名/全量刷新 -│ ├── DBMetaContext.java # 元数据缓存（单例，key=jdbcKey） -│ ├── DbTable.java # 表元数据（列、主键） -│ ├── DbColumn.java # 列元数据（类型、是否主键） -│ └── JavaDBTypeConvertor.java # JDBC 类型→Java 类型映射 -├── supporter/ -│ ├── DBEventSupporter.java # 表级事件支持判断接口 -│ └── DefaultDBEventSupporter.java # 默认实现（始终支持） -├── query/ -│ └── JdbcQuery.java # 通过真实 Connection 执行 SELECT 的工具 -├── serializable/ -│ └── DBTableSerializableHelper.java # DbTable Java 序列化至 .dbstream/ 缓存目录，支持按表名清除 -└── utils/ - ├── SQLUtils.java # SQL 分类判断（isInsert/isUpdate/isDelete） - ├── JdbcPropertyUtils.java # jdbcKey 生成：sha256(jdbcUrl#schema) - ├── SHA256Utils.java # SHA-256 哈希 - ├── VersionUtils.java # 从 MANIFEST.MF 读取版本号 - ├── FileReaderUtils.java # 读取 .key 文件（手动主键配置） - └── ResultSetUtils.java # 结果集判断工具 -``` - -## 核心功能 - -### 1. JDBC 代理驱动 - -通过 Java SPI（`META-INF/services/java.sql.Driver`）注册 `DBStreamProxyDriver`。`connect()` 时自动发现真实驱动、扫描数据库元数据，返回 `ConnectionProxy`。所有 Statement 创建均被代理。 - -### 2. SQL 解析与事件生成 - -基于正则的 SQL 解析器（`InsertSQLParser`/`UpdateSQLParser`/`DeleteSQLParser`）提取表名、列、WHERE 子句。`DBEventParser` 在 `before()` 阶段执行 prepare（UPDATE/DELETE 查询 pre-image），在 `after()` 阶段组装 `DBEvent` 对象。 - -### 3. 事务级事件管理 - -`TransactionEventPools` 使用 ThreadLocal 累积事件。auto-commit 模式下立即推送；手动事务下在 `commit()` 时批量推送，`rollback()` 时丢弃。 - -### 4. 数据库元数据扫描与缓存 - -`DBScanner` 通过 `java.sql.DatabaseMetaData` 扫描表结构，序列化到 `.dbstream//` 目录缓存。支持 `.key` 文件手动指定主键。提供 `scanTable(tableName)` 按单表扫描、`scanAllTables()` 全量扫描接口。 - -### 5. 元数据动态刷新 - -`DBMetaData` 支持运行时按表名（`refreshTable`）或全量（`refreshAll`）刷新元数据，适用于动态 DDL 场景。刷新时先清除序列化缓存再重新扫描，更新内存中已有表数据并追加新表。通过 `DBStreamContext` 暴露公共 API，接受标准 `java.sql.Connection`。 - -## 对外 API - -### DBStreamContext（`com.codingapi.dbstream.DBStreamContext`） - -全局单例，所有公共能力的统一入口。 - -| 方法 | 说明 | -|------|------| -| `getInstance()` | 获取单例 | -| `addListener(SQLExecuteListener)` | 注册 SQL 执行监听器 | -| `cleanCustomListeners()` | 清除自定义监听器（保留系统监听器） | -| `addEventPusher(DBEventPusher)` | 注册事件推送者 | -| `cleanEventPushers()` | 清除所有推送者 | -| `setDbEventSupporter(DBEventSupporter)` | 设置表级事件支持过滤器 | -| `getMetaData(jdbcKey)` | 获取指定数据源元数据 | -| `loadDbKeys()` | 列出所有已注册数据源 key | -| `metaDataList()` | 返回全部数据源元数据列表 | -| `clearAll()` / `clear(jdbcKey)` | 清除元数据缓存 | -| `refreshTable(Connection, jdbcKey, tableName)` | 刷新指定表的元数据（清除缓存后重新扫描） | -| `refreshAll(Connection, jdbcKey)` | 全量刷新指定数据源的元数据 | - -### SQLExecuteListener（`com.codingapi.dbstream.listener.SQLExecuteListener`） - -SQL 执行监听接口，按 `order()` 排序执行。 - -| 方法 | 说明 | -|------|------| -| `order()` | 执行顺序，越小越靠前 | -| `before(SQLRunningState)` | SQL 执行前回调 | -| `after(SQLRunningState, Object)` | SQL 执行后回调 | - -### DBEventPusher（`com.codingapi.dbstream.event.DBEventPusher`） - -事件推送接口，事务提交时调用。 - -| 方法 | 说明 | -|------|------| -| `push(JdbcQuery, List)` | 推送事件列表 | - -### DBEventSupporter（`com.codingapi.dbstream.supporter.DBEventSupporter`） - -表级事件过滤接口。 - -| 方法 | 说明 | -|------|------| -| `support(Properties, DbTable)` | 判断是否对该表启用事件捕获 | - -## 模块规范 - -- Java 8 兼容，不使用高于 Java 8 的 API -- 单例模式统一使用 `private static final` 饿汉初始化 -- SQL 解析基于正则，不引入外部解析库 -- 并发控制：ThreadLocal 隔离事务状态，ConcurrentHashMap 缓存驱动/元数据，CopyOnWriteArrayList 管理监听器 -- Lombok 仅用于编译期，标记为 optional，不传递给下游 - -## 构建指令 - -```bash -./mvnw compile -./mvnw test -./mvnw package -./mvnw clean deploy -P ossrh -``` - -## 变更说明 - -- **2026-04-28**: 全量同步最新代码。版本号更新为 1.0.18；新增「元数据动态刷新」核心功能章节；DBStreamContext API 表补充 `metaDataList()`、`refreshTable()`、`refreshAll()`；项目结构注释同步更新 diff --git a/docs/changes/819340-dynamic-table-metadata-refresh.md b/docs/changes/819340-dynamic-table-metadata-refresh.md deleted file mode 100644 index 05a7b64..0000000 --- a/docs/changes/819340-dynamic-table-metadata-refresh.md +++ /dev/null @@ -1,42 +0,0 @@ -# 动态表元数据刷新机制 - -> 编码: 819340 | 日期: 2026-04-28 | 调整类型: 功能增强 | 状态: 草稿 - -## 调整背景 - -运行时动态创建或修改的表，无法主动的加载更新到缓存数据DBMetaData。当前 DBMetaData 在首次连接时通过 `DBScanner` 扫描全部表结构并缓存到 `DBMetaContext`（ConcurrentHashMap），后续不再主动更新。这意味着： - -1. 应用运行期间通过 DDL（CREATE TABLE / ALTER TABLE）新增或修改的表结构不会反映到缓存中 -2. 对新增表的 INSERT/UPDATE/DELETE 操作无法触发 CDC 事件（因为 `DbTable` 查找返回 null） -3. 修改了表的列结构（如新增列、修改主键）后，事件捕获仍使用旧的元数据，可能导致数据丢失或解析错误 -4. 当前仅支持通过 `DBMetaData.addUpdateSubscribe()` 注册订阅表名，在下次 SQL 执行到该表时触发局部更新，但这属于被动机制，需要预先知道表名 - -## 关键要点 - -- **涉及模块**: dbstream-driver - - `scanner/DBScanner`（元数据扫描入口） - - `scanner/DBMetaContext`（元数据缓存单例） - - `scanner/DBMetaData`（单数据源元数据，含 `addUpdateSubscribe` 订阅机制） - - `proxy/ConnectionProxy`（连接代理，元数据加载触发点） - - `DBStreamContext`（对外 API，`clear`/`clearAll` 可强制刷新缓存） - -- **影响范围**: - - 所有依赖 `DBMetaData.getTable()` 获取表信息的调用方 - - `DBEventListener.before()` 中查找 `DbTable` 的逻辑 - - `DBTableSerializableHelper` 的本地文件缓存一致性 - - 现有的 `addUpdateSubscribe` 被动订阅机制可能需要重新评估 - -- **现有缓解机制**: - - `DBStreamContext.clear(jdbcKey)` / `clearAll()` 可手动清除缓存，下次连接时自动重新扫描 - - `DBMetaData.addUpdateSubscribe(tableName)` 支持注册待更新表名，在 SQL 执行到该表时被动触发 `DBScanner.findTableMetadata()` 局部更新 - - `SQLRunningState.triggerDBMetaData(tableName)` 在 `DBEventListener.before()` 中被调用，是被动更新的触发点 - -- **可能的改进方向**: - - 提供主动刷新 API（按表名或全量刷新） - - 支持 DDL 语句检测（拦截 CREATE/ALTER TABLE 自动触发元数据刷新） - - 定时刷新机制（可选） - - 事件驱动的刷新回调 - -## 备注 - -{待人工补充} diff --git a/docs/components/dbstream-driver_DBEventPusher.md b/docs/components/dbstream-driver_DBEventPusher.md deleted file mode 100644 index 157cea5..0000000 --- a/docs/components/dbstream-driver_DBEventPusher.md +++ /dev/null @@ -1,187 +0,0 @@ ---- -module: dbstream-driver -name: DBEventPusher -description: 数据变更事件推送接口，在事务提交时接收 DBEvent 列表并推送到外部系统（消息队列、日志、缓存同步等）。 ---- - -# DBEventPusher - -- **来源**: 自有 -- **所属 module**: dbstream-driver -- **Maven 坐标**: com.codingapi.dbstream:dbstream-driver:1.0.18 - -## 何时使用 - -需要将数据库变更事件推送到外部系统时使用，例如：同步数据到 Elasticsearch/Redis、发送变更通知到消息队列、构建审计日志、实现跨服务数据一致性。未注册任何 `DBEventPusher` 时，事件默认输出到 stdout。 - -## 如何引用 - -### Maven 坐标 - -```xml - - com.codingapi.dbstream - dbstream-driver - 1.0.18 - -``` - -## API 说明 - -### 核心类 - -| 类名 | 包路径 | 说明 | -|------|--------|------| -| `DBEventPusher` | `com.codingapi.dbstream.event` | 事件推送接口，业务方实现此接口 | -| `DefaultDBEventPusher` | `com.codingapi.dbstream.event` | 默认实现，将事件打印到 stdout | -| `DBEventContext` | `com.codingapi.dbstream.event` | 推送者注册中心（单例），管理推送者列表并分发事件 | -| `DBEvent` | `com.codingapi.dbstream.event` | 数据变更事件对象（表名、类型、数据、主键、事务标识） | -| `EventType` | `com.codingapi.dbstream.event` | 事件类型枚举：INSERT / UPDATE / DELETE | -| `TransactionEventPools` | `com.codingapi.dbstream.event` | 事务级事件池（ThreadLocal），管理事件的累积与提交/回滚 | -| `JdbcQuery` | `com.codingapi.dbstream.query` | JDBC 查询工具，在 push 回调中可用来查询补充数据 | - -### 关键方法 - -#### DBEventPusher 接口 - -| 方法签名 | 参数说明 | 返回值 | 说明 | -|----------|----------|--------|------| -| `push(JdbcQuery, List)` | jdbcQuery 可用于查询补充数据；events 为本次事务产生的变更事件列表 | void | 事务提交时调用。auto-commit 模式下每条 SQL 执行后立即调用；手动事务下在 commit() 时批量调用 | - -#### DBEvent 数据访问方法 - -| 方法签名 | 返回值 | 说明 | -|----------|--------|------| -| `getTableName()` | String | 变更的数据表名称 | -| `getType()` | EventType | 事件类型：INSERT / UPDATE / DELETE | -| `isInsert()` / `isUpdate()` / `isDelete()` | boolean | 便捷类型判断 | -| `getData()` | Map\ | 变更数据（INSERT=新数据，UPDATE=更新后数据，DELETE=删除前数据） | -| `set(String key, Object value)` | void | 向 data Map 中添加键值对 | -| `getPrimaryKeys()` | List\ | 主键列名列表 | -| `addPrimaryKey(String)` | void | 添加主键列名（去重） | -| `hasPrimaryKeys()` | boolean | 是否有主键信息 | -| `getTransactionKey()` | String | 所属事务标识 | -| `getJdbcKey()` | String | 数据源唯一标识 | -| `getJdbcUrl()` | String | 数据库连接 URL | -| `getTimestamp()` | long | 事件创建时间戳 | -| `getPushTimestamp()` | long | 事件推送时间戳 | - -#### EventType 枚举值 - -| 值 | 说明 | -|----|------| -| `INSERT` | 插入事件 | -| `UPDATE` | 更新事件 | -| `DELETE` | 删除事件 | - -#### DBEventContext 管理方法 - -| 方法签名 | 返回值 | 说明 | -|----------|--------|------| -| `DBEventContext.getInstance()` | DBEventContext | 获取单例 | -| `addPusher(DBEventPusher)` | void | 注册推送者 | -| `clean()` | void | 清除所有已注册的推送者 | - -#### TransactionEventPools 关键方法 - -| 方法签名 | 返回值 | 说明 | -|----------|--------|------| -| `isAutoCommit()` | boolean | 当前是否自动提交模式 | -| `setAutoCommit(boolean)` | void | 设置自动提交模式 | -| `addEvents(JdbcQuery, String, List)` | void | 添加事件，auto-commit 模式下立即推送 | -| `commitEvents(JdbcQuery, String)` | void | 设置事务标识并推送全部累积事件 | -| `rollbackEvents(JdbcQuery, String)` | void | 回滚，丢弃所有累积事件 | -| `clear()` | void | 清除事件池 ThreadLocal | -| `reset()` | void | 同时清除事件池和自动提交模式 ThreadLocal | - -### 事件数据说明 - -| EventType | data 内容 | 说明 | -|-----------|----------|------| -| INSERT | 新插入行的列值 | 主键通过 `getStatementGenerateKeys()` 或 SQL 解析获取 | -| UPDATE | 更新后的行数据（post-image） | UPDATE 事件在 before 阶段查询 pre-image，after 阶段查询 post-image | -| DELETE | 删除前的行数据（pre-image） | 在 before 阶段通过 WHERE 条件查询当前数据 | - -### 配置项 - -本组件无独立配置项。推送者通过 `DBStreamContext.addEventPusher()` 代码注册。 - -## 使用示例 - -### 基础用法：注册事件推送者 - -```java -DBStreamContext ctx = DBStreamContext.getInstance(); - -ctx.addEventPusher((jdbcQuery, events) -> { - for (DBEvent event : events) { - String action = event.isInsert() ? "新增" : - event.isUpdate() ? "修改" : "删除"; - System.out.println(action + " 表:" + event.getTableName()); - System.out.println("主键:" + event.getPrimaryKeys()); - System.out.println("数据:" + event.getData()); - System.out.println("事务:" + event.getTransactionKey()); - } -}); -``` - -### 推送到消息队列 - -```java -ctx.addEventPusher((jdbcQuery, events) -> { - for (DBEvent event : events) { - Map message = new HashMap<>(); - message.put("table", event.getTableName()); - message.put("type", event.getType().name()); - message.put("data", event.getData()); - message.put("primaryKeys", event.getPrimaryKeys()); - message.put("transactionKey", event.getTransactionKey()); - message.put("timestamp", event.getTimestamp()); - // 发送到 Kafka/RabbitMQ 等 - mqProducer.send(JSON.toJSONString(message)); - } -}); -``` - -### 在 push 中查询补充数据 - -```java -ctx.addEventPusher((jdbcQuery, events) -> { - for (DBEvent event : events) { - if (event.isUpdate() && "t_order".equals(event.getTableName())) { - // 利用 JdbcQuery 查询关联数据 - List> extra = jdbcQuery.query( - "SELECT * FROM t_order_item WHERE order_id = ?", - Collections.singletonList(event.getData().get("id")) - ); - // 合并发送 - } - } -}); -``` - -### 多推送者注册 - -```java -// 可注册多个推送者，全部都会被调用 -ctx.addEventPusher(new ElasticsearchSyncPusher()); -ctx.addEventPusher(new AuditLogPusher()); -ctx.addEventPusher(new CacheInvalidationPusher()); - -// 清除所有推送者 -ctx.cleanEventPushers(); -``` - -## 注意事项 - -- **事务边界**: auto-commit 模式下每条 SQL 执行后立即触发 `push()`；手动事务模式下事件在 `commit()` 时批量推送，`rollback()` 时丢弃。推送者实现不需要关心事务边界 -- **事务标识**: 每个 `DBEvent` 的 `transactionKey` 标识其所属事务，同一事务内的多个事件共享相同的 transactionKey，可用于保证消费端的顺序性和原子性 -- **多推送者**: 支持注册多个 `DBEventPusher`，所有注册者都会收到全部事件。使用 `CopyOnWriteArrayList` 存储，支持运行时注册 -- **默认行为**: 未注册任何推送者时，`DefaultDBEventPusher` 将事件打印到 stdout。注册至少一个推送者后，默认行为不再触发 -- **线程安全**: `TransactionEventPools` 使用 ThreadLocal 隔离事务状态，推送者实现无需考虑多线程竞争。但 `push()` 回调内应避免耗时操作，以免阻塞数据库连接 -- **JdbcQuery 使用**: `push()` 回调中的 `JdbcQuery` 参数使用真实（非代理）Connection 执行查询，不会触发事件递归 -- **push 异常**: `push()` 中抛出异常会影响后续推送者的调用。建议在推送者内部做异常捕获和容错处理，避免影响其他推送者 - -## 变更说明 - -- **2026-04-28**: 同步最新代码，新增 `DBEvent.set()`、`DBEvent.addPrimaryKey()` 方法文档；新增 `TransactionEventPools.clear()`、`TransactionEventPools.reset()` 方法文档；Maven 坐标版本号更新为 1.0.18 diff --git a/docs/components/dbstream-driver_DBEventSupporter.md b/docs/components/dbstream-driver_DBEventSupporter.md deleted file mode 100644 index cba5146..0000000 --- a/docs/components/dbstream-driver_DBEventSupporter.md +++ /dev/null @@ -1,161 +0,0 @@ ---- -module: dbstream-driver -name: DBEventSupporter -description: 表级事件过滤接口，用于控制哪些数据库表启用数据变更事件捕获，默认全部支持。 ---- - -# DBEventSupporter - -- **来源**: 自有 -- **所属 module**: dbstream-driver -- **Maven 坐标**: com.codingapi.dbstream:dbstream-driver:1.0.18 - -## 何时使用 - -需要精确控制哪些数据库表参与 CDC 事件捕获时使用。默认情况下（`DefaultDBEventSupporter`）所有具备主键的表都会被捕获。当只关心部分核心业务表的变更（如订单、支付、库存），或需要排除日志表、临时表等高频低价值表时，实现此接口进行过滤。 - -## 如何引用 - -### Maven 坐标 - -```xml - - com.codingapi.dbstream - dbstream-driver - 1.0.18 - -``` - -## API 说明 - -### 核心类 - -| 类名 | 包路径 | 说明 | -|------|--------|------| -| `DBEventSupporter` | `com.codingapi.dbstream.supporter` | 表级事件过滤接口 | -| `DefaultDBEventSupporter` | `com.codingapi.dbstream.supporter` | 默认实现，始终返回 true | -| `DBStreamContext` | `com.codingapi.dbstream` | 全局入口，通过 `setDbEventSupporter()` 注入自定义实现 | -| `DbTable` | `com.codingapi.dbstream.scanner` | 表元数据（表名、列、主键），过滤判断的输入参数 | -| `DbColumn` | `com.codingapi.dbstream.scanner` | 列元数据（列名、类型、Java 类型、是否可空、注释、长度、是否主键） | - -### 关键方法 - -#### DBEventSupporter 接口 - -| 方法签名 | 参数说明 | 返回值 | 说明 | -|----------|----------|--------|------| -| `support(Properties, DbTable)` | info=数据库连接属性；dbTable=表元数据 | boolean | 返回 true 表示对该表启用事件捕获，false 表示跳过 | - -#### DbTable 判断方法（用于过滤逻辑） - -| 方法签名 | 返回值 | 说明 | -|----------|--------|------| -| `getName()` | String | 表名称 | -| `getComment()` | String | 表注释 | -| `getColumns()` | List\ | 全部列元数据 | -| `getPrimaryKeys()` | List\ | 主键列名列表 | -| `getPrimaryColumns()` | List\ | 主键列的完整 DbColumn 对象列表 | -| `isPrimaryKey(String columnName)` | boolean | 判断指定列名是否为主键 | -| `getColumnByName(String column)` | DbColumn | 根据列名查找 DbColumn 对象 | -| `hasColumns()` | boolean | 是否有列定义 | -| `hasPrimaryKeys()` | boolean | 是否有主键 | - -#### DbColumn 字段 - -| 字段 | 类型 | 说明 | -|------|------|------| -| `name` | String | 列名称 | -| `type` | String | 数据库字段类型 | -| `javaType` | Class\ | Java 对应类型 | -| `nullable` | boolean | 是否可为空 | -| `comment` | String | 列注释 | -| `size` | int | 字段长度 | -| `primaryKey` | boolean | 是否为主键（由 DbTable 运行时回填） | - -#### 注册方式 - -通过 `DBStreamContext` 设置： - -| 方法签名 | 说明 | -|----------|------| -| `DBStreamContext.getInstance().setDbEventSupporter(DBEventSupporter)` | 设置自定义过滤器 | -| `DBStreamContext.getInstance().support(Properties, DbTable)` | 内部调用，先检查表是否有列和主键，再委托给 supporter | - -### 过滤执行时机 - -`DBEventSupporter.support()` 在每次 SQL 执行前的 `DBEventListener.before()` 中被调用，流程为： - -1. SQL 解析器提取表名 -2. 从 `DBMetaData` 查找 `DbTable` -3. 检查表是否有列和主键（前置条件） -4. 调用 `DBEventSupporter.support(properties, dbTable)` 判断 -5. 返回 false → 跳过该表的事件解析，无性能开销 - -### 配置项 - -本组件无独立配置项。通过 `DBStreamContext.setDbEventSupporter()` 代码注入。 - -## 使用示例 - -### 基础用法：按表名前缀过滤 - -```java -DBStreamContext.getInstance().setDbEventSupporter((info, dbTable) -> { - // 仅捕获 t_ 前缀的业务表 - return dbTable.getName().startsWith("t_"); -}); -``` - -### 按表名白名单过滤 - -```java -Set targetTables = new HashSet<>(Arrays.asList( - "t_order", "t_order_item", "t_payment", "t_inventory" -)); - -DBStreamContext.getInstance().setDbEventSupporter((info, dbTable) -> { - return targetTables.contains(dbTable.getName().toLowerCase()); -}); -``` - -### 按数据源过滤 - -```java -DBStreamContext.getInstance().setDbEventSupporter((info, dbTable) -> { - String jdbcUrl = info.getProperty("jdbc.url"); - // 只对订单库启用事件捕获 - if (jdbcUrl != null && jdbcUrl.contains("order-db")) { - return dbTable.getName().startsWith("t_"); - } - return false; -}); -``` - -### 结合表结构过滤 - -```java -DBStreamContext.getInstance().setDbEventSupporter((info, dbTable) -> { - // 排除没有注释的表（通常是临时表或内部表） - if (dbTable.getComment() == null || dbTable.getComment().isEmpty()) { - return false; - } - // 排除日志类表 - if (dbTable.getName().toUpperCase().contains("LOG")) { - return false; - } - return true; -}); -``` - -## 注意事项 - -- **前置条件**: `DBStreamContext.support()` 在调用 `DBEventSupporter` 之前会先检查表是否有列和主键。无主键的表（如日志表、关联表）不会被捕获，即使 supporter 返回 true -- **设置时机**: 建议在应用启动阶段一次性设置，`setDbEventSupporter()` 不是线程安全的 -- **默认行为**: 未设置自定义 supporter 时，使用 `DefaultDBEventSupporter`（始终返回 true）。设置后会覆盖默认实现 -- **性能影响**: `support()` 在每次 INSERT/UPDATE/DELETE 执行前被调用，实现应保持轻量，避免 IO 操作或复杂计算 -- **null 安全**: `support()` 的两个参数不会为 null（在 `DBStreamContext.support()` 中已做前置校验） -- **单实例**: 全局只能设置一个 `DBEventSupporter`，后设置的会覆盖先前的。如需组合多个过滤条件，在实现内部组合 - -## 变更说明 - -- **2026-04-28**: 同步最新代码，新增 `DbTable.getPrimaryColumns()`、`isPrimaryKey()`、`getColumnByName()` 方法文档；补充 `DbColumn` 完整 7 字段文档（含 `javaType`、`size`、`primaryKey`）；Maven 坐标版本号更新为 1.0.18 diff --git a/docs/components/dbstream-driver_DBStreamContext.md b/docs/components/dbstream-driver_DBStreamContext.md deleted file mode 100644 index ded109c..0000000 --- a/docs/components/dbstream-driver_DBStreamContext.md +++ /dev/null @@ -1,183 +0,0 @@ ---- -module: dbstream-driver -name: DBStreamContext -description: DBStream 的全局统一入口，用于注册 SQL 执行监听器、事件推送者、表级过滤器，以及管理数据库元数据缓存。 ---- - -# DBStreamContext - -- **来源**: 自有 -- **所属 module**: dbstream-driver -- **Maven 坐标**: com.codingapi.dbstream:dbstream-driver:1.0.18 - -## 何时使用 - -当需要在应用中集成 DBStream 的 CDC 能力时，通过 `DBStreamContext` 单例完成所有配置：注册自定义的 SQL 执行监听器、设置数据变更事件推送目标、过滤需要捕获事件的表，以及查询和管理数据库元数据。 - -## 如何引用 - -### Maven 坐标 - -```xml - - com.codingapi.dbstream - dbstream-driver - 1.0.18 - -``` - -## API 说明 - -### 核心类 - -| 类名 | 包路径 | 说明 | -|------|--------|------| -| `DBStreamContext` | `com.codingapi.dbstream` | 全局单例入口，聚合所有公共 API | -| `SQLExecuteListener` | `com.codingapi.dbstream.listener` | SQL 执行监听接口（before/after） | -| `SQLRunningContext` | `com.codingapi.dbstream.listener` | 监听器注册中心，按 order 编排回调 | -| `SQLRunningState` | `com.codingapi.dbstream.listener` | 单次 SQL 执行上下文（SQL、参数、元数据） | -| `DBEventPusher` | `com.codingapi.dbstream.event` | 数据变更事件推送接口 | -| `DBEventContext` | `com.codingapi.dbstream.event` | 推送者注册与事件分发 | -| `DBEvent` | `com.codingapi.dbstream.event` | 数据变更事件（表名、类型、数据、主键、事务标识） | -| `DBEventSupporter` | `com.codingapi.dbstream.supporter` | 表级事件过滤接口 | -| `DBMetaData` | `com.codingapi.dbstream.scanner` | 单数据源全部表元数据 | -| `DBMetaContext` | `com.codingapi.dbstream.scanner` | 元数据缓存单例（key=jdbcKey） | - -### 关键方法 - -#### 监听器管理 - -| 方法签名 | 参数说明 | 返回值 | 说明 | -|----------|----------|--------|------| -| `addListener(SQLExecuteListener)` | SQL 执行监听器 | void | 注册自定义监听器，按 `order()` 排序执行 | -| `cleanCustomListeners()` | 无 | void | 清除所有自定义监听器，保留三个系统监听器（Insert/Update/Delete） | - -#### 事件推送管理 - -| 方法签名 | 参数说明 | 返回值 | 说明 | -|----------|----------|--------|------| -| `addEventPusher(DBEventPusher)` | 事件推送实现 | void | 注册推送者，事务提交时回调 `push()` | -| `cleanEventPushers()` | 无 | void | 清除所有已注册的推送者 | - -#### 事件过滤 - -| 方法签名 | 参数说明 | 返回值 | 说明 | -|----------|----------|--------|------| -| `setDbEventSupporter(DBEventSupporter)` | 自定义过滤器 | void | 设置表级事件支持判断，默认全部支持 | -| `support(Properties, DbTable)` | 连接属性、表元数据 | boolean | 判断是否对该表启用事件捕获（要求表有列和主键） | - -#### 元数据查询 - -| 方法签名 | 参数说明 | 返回值 | 说明 | -|----------|----------|--------|------| -| `getMetaData(String)` | jdbcKey | DBMetaData | 获取指定数据源的元数据 | -| `loadDbKeys()` | 无 | List\ | 列出所有已注册数据源的 key | -| `metaDataList()` | 无 | List\ | 返回全部数据源元数据列表 | -| `clearAll()` | 无 | void | 清除所有元数据缓存，下次访问时重新扫描 | -| `clear(String)` | jdbcKey | void | 清除指定数据源的元数据缓存 | - -#### 元数据刷新 - -| 方法签名 | 参数说明 | 返回值 | 说明 | -|----------|----------|--------|------| -| `refreshTable(Connection, String, String)` | 数据库连接、jdbcKey、表名 | void | 刷新指定表的元数据。适用于运行时动态创建或修改表结构后手动触发更新。会清除该表的序列化缓存并重新扫描 | -| `refreshAll(Connection, String)` | 数据库连接、jdbcKey | void | 全量刷新指定数据源的元数据。重新扫描所有表结构并更新缓存 | - -### 配置项 - -| 配置Key | 类型 | 默认值 | 说明 | -|---------|------|--------|------| -| `jdbc.url` | String（Properties 内） | 无 | 数据库连接 URL，自动写入 Properties | -| `jdbc.key` | String（Properties 内） | sha256(jdbcUrl#schema) | 数据源唯一标识，自动生成 | - -## 使用示例 - -### 基础用法：注册事件推送者 - -```java -DBStreamContext ctx = DBStreamContext.getInstance(); - -// 注册事件推送者（事务提交时收到变更事件） -ctx.addEventPusher((jdbcQuery, events) -> { - for (DBEvent event : events) { - System.out.println(event.getType() + " on " + event.getTableName()); - System.out.println("data: " + event.getData()); - System.out.println("primaryKeys: " + event.getPrimaryKeys()); - } -}); - -// 之后正常使用 JDBC/JPA 操作数据库即可自动捕获事件 -``` - -### 注册 SQL 执行监听器 - -```java -ctx.addListener(new SQLExecuteListener() { - @Override - public int order() { - return 0; // 越小越先执行 - } - - @Override - public void before(SQLRunningState state) { - System.out.println("执行 SQL: " + state.getSql()); - System.out.println("参数: " + state.getListParams()); - } - - @Override - public void after(SQLRunningState state, Object result) { - System.out.println("执行耗时: " + state.getExecuteTimestamp()); - } -}); -``` - -### 过滤需要捕获的表 - -```java -ctx.setDbEventSupporter((info, dbTable) -> { - // 仅捕获 t_order 开头的表 - return dbTable.getName().startsWith("t_order"); -}); -``` - -### 清理与重置 - -```java -// 清除自定义监听器（保留系统级 Insert/Update/Delete 监听器） -ctx.cleanCustomListeners(); - -// 清除所有事件推送者 -ctx.cleanEventPushers(); - -// 清除元数据缓存（下次连接时自动重新扫描） -ctx.clearAll(); -``` - -### 动态刷新元数据 - -```java -// 运行时创建了新表或修改了表结构后，手动刷新元数据 -try (Connection conn = dataSource.getConnection()) { - String jdbcKey = ctx.loadDbKeys().get(0); - - // 刷新单张表 - ctx.refreshTable(conn, jdbcKey, "t_new_table"); - - // 或全量刷新所有表 - ctx.refreshAll(conn, jdbcKey); -} -``` - -## 注意事项 - -- **单例模式**: `DBStreamContext` 使用饿汉单例（`private static final`），全局唯一实例通过 `getInstance()` 获取 -- **线程安全**: 监听器列表使用 `CopyOnWriteArrayList`，元数据缓存使用 `ConcurrentHashMap`，支持并发注册和查询；但 `setDbEventSupporter()` 不是线程安全的，建议在应用启动阶段一次性设置 -- **系统监听器不可移除**: `cleanCustomListeners()` 仅移除自定义监听器，InsertEventListener / UpdateEventListener / DeleteEventListener 三个系统监听器始终保留 -- **默认推送行为**: 未注册任何 `DBEventPusher` 时，事件输出到 stdout（`DefaultDBEventPusher`） -- **元数据自动加载**: 清除元数据缓存后，下次数据库连接时会自动重新扫描表结构，无需手动触发 -- **事件过滤前置条件**: 即使 `DBEventSupporter` 返回 true，表仍须具备列定义和主键才会被纳入事件捕获 -- **元数据刷新**: `refreshTable()` 和 `refreshAll()` 接受标准 `java.sql.Connection`（不需要是 ConnectionProxy），适用于运行时动态 DDL 场景。`refreshTable()` 会先清除该表的序列化缓存文件再重新扫描 - -## 变更说明 - -- **2026-04-28**: 同步最新代码，新增 `refreshTable(Connection, String, String)` 和 `refreshAll(Connection, String)` 元数据刷新 API 文档及使用示例；Maven 坐标版本号更新为 1.0.18 diff --git a/docs/components/dbstream-driver_SQLExecuteListener.md b/docs/components/dbstream-driver_SQLExecuteListener.md deleted file mode 100644 index 9d7b425..0000000 --- a/docs/components/dbstream-driver_SQLExecuteListener.md +++ /dev/null @@ -1,168 +0,0 @@ ---- -module: dbstream-driver -name: SQLExecuteListener -description: SQL 执行拦截监听接口，允许在 SQL 执行前后插入自定义逻辑，如日志记录、性能监控、审计追踪等。 ---- - -# SQLExecuteListener - -- **来源**: 自有 -- **所属 module**: dbstream-driver -- **Maven 坐标**: com.codingapi.dbstream:dbstream-driver:1.0.18 - -## 何时使用 - -需要在 SQL 执行前后插入自定义处理逻辑时使用，例如：SQL 审计日志、慢查询监控、SQL 改写、性能统计。DBStream 内部的三个系统监听器（Insert/Update/Delete）也基于此接口实现数据变更事件捕获。 - -## 如何引用 - -### Maven 坐标 - -```xml - - com.codingapi.dbstream - dbstream-driver - 1.0.18 - -``` - -## API 说明 - -### 核心类 - -| 类名 | 包路径 | 说明 | -|------|--------|------| -| `SQLExecuteListener` | `com.codingapi.dbstream.listener` | SQL 执行监听接口，自定义监听器的契约 | -| `SQLRunningContext` | `com.codingapi.dbstream.listener` | 监听器注册中心（单例），按 order 编排 before/after 回调 | -| `SQLRunningState` | `com.codingapi.dbstream.listener` | 单次 SQL 执行的完整上下文（SQL、参数、元数据、执行耗时） | -| `SQLRunningParam` | `com.codingapi.dbstream.listener` | 参数追踪器，支持 index 和 key 两种参数绑定模式 | -| `DBEventListener` | `com.codingapi.dbstream.listener.dbevent` | 抽象基类，实现 `SQLExecuteListener`，串联 SQL 解析→事件生成→事务池 | -| `InsertEventListener` | `com.codingapi.dbstream.listener.dbevent` | INSERT 系统监听器（order=100） | -| `UpdateEventListener` | `com.codingapi.dbstream.listener.dbevent` | UPDATE 系统监听器（order=100） | -| `DeleteEventListener` | `com.codingapi.dbstream.listener.dbevent` | DELETE 系统监听器（order=100） | -| `JdbcQuery` | `com.codingapi.dbstream.query` | 通过真实 Connection 执行 SELECT 查询的工具，在 `SQLRunningState` 中暴露 | - -### 关键方法 - -#### SQLExecuteListener 接口方法 - -| 方法签名 | 参数说明 | 返回值 | 说明 | -|----------|----------|--------|------| -| `order()` | 无 | int | 执行顺序，值越小越先执行。系统监听器 order=100，自定义监听器建议用更小的值以优先执行 | -| `before(SQLRunningState)` | SQL 执行上下文 | void | SQL 执行前回调。可读取 SQL、参数、元数据；抛出 SQLException 可阻止执行 | -| `after(SQLRunningState, Object)` | SQL 执行上下文 + 执行结果 | void | SQL 执行后回调。result 通常是 int（影响行数）或 ResultSet | - -#### SQLRunningState 关键方法 - -| 方法签名 | 返回值 | 说明 | -|----------|--------|------| -| `getSql()` | String | 当前执行的 SQL 语句 | -| `getListParams()` | List\ | 按索引排序的参数值列表 | -| `getMetaData()` | DBMetaData | 当前数据源的元数据 | -| `getDbTable(String)` | DbTable | 按表名查找元数据 | -| `getTransactionKey()` | String | 当前事务标识 | -| `getJdbcUrl()` | String | 当前数据库连接 URL | -| `getJdbcKey()` | String | 当前数据源唯一标识 | -| `getDriverProperties()` | Properties | 驱动配置信息 | -| `getExecuteTimestamp()` | long | SQL 执行耗时（毫秒，after 减 before） | -| `getBeginTimestamp()` | long | 执行开始时间戳 | -| `getAfterTimestamp()` | long | 执行结束时间戳 | -| `isJdbcBatchMode()` | boolean | 是否处于 JDBC 批量模式 | -| `getBatchSQLRunningStateList()` | List\ | 批量模式下获取每条 SQL 的独立上下文列表 | -| `getStatement()` | Statement | 获取底层 Statement 对象 | -| `getConnectionProxy()` | ConnectionProxy | 获取代理连接对象 | -| `getJdbcQuery()` | JdbcQuery | 获取 JDBC 查询工具 | -| `query(String)` | List\\> | 执行 SELECT 查询（通过真实连接） | -| `query(String, List\)` | List\\> | 带参数的 SELECT 查询 | -| `getStatementGenerateKeys(DbTable)` | List\\> | 获取自增主键生成结果 | -| `getResult()` | Object | SQL 执行结果 | -| `triggerDBMetaData(String tableName)` | void | 触发指定表的元数据订阅更新 | - -#### SQLRunningContext 管理方法 - -| 方法签名 | 返回值 | 说明 | -|----------|--------|------| -| `SQLRunningContext.getInstance()` | SQLRunningContext | 获取单例 | -| `getListeners()` | List\ | 获取已注册的监听器列表 | -| `addListener(SQLExecuteListener)` | void | 注册监听器（去重 + 按 order 排序） | -| `addSystemListeners()` | void | 注册系统默认监听器（Insert/Update/Delete） | -| `cleanCustomListeners()` | void | 移除自定义监听器，保留三个系统监听器 | -| `before(SQLRunningState)` | void | 触发所有监听器的 before 回调 | -| `after(SQLRunningState, Object)` | void | 触发所有监听器的 after 回调 | - -### 配置项 - -本组件无独立配置项。监听器通过 `DBStreamContext.addListener()` 代码注册。 - -## 使用示例 - -### 基础用法：SQL 审计日志 - -```java -DBStreamContext.getInstance().addListener(new SQLExecuteListener() { - @Override - public int order() { - return 0; // 优先于系统监听器（order=100）执行 - } - - @Override - public void before(SQLRunningState state) { - System.out.println("执行 SQL: " + state.getSql()); - System.out.println("参数: " + state.getListParams()); - } - - @Override - public void after(SQLRunningState state, Object result) { - System.out.println("执行耗时: " + state.getExecuteTimestamp() + "ms"); - } -}); -``` - -### 慢查询监控 - -```java -DBStreamContext.getInstance().addListener(new SQLExecuteListener() { - @Override - public int order() { return 0; } - - @Override - public void before(SQLRunningState state) {} - - @Override - public void after(SQLRunningState state, Object result) { - long cost = state.getExecuteTimestamp(); - if (cost > 1000) { - log.warn("慢查询 [{}ms]: {} params={}", cost, state.getSql(), state.getListParams()); - } - } -}); -``` - -### 在 before 中执行辅助查询 - -```java -@Override -public void before(SQLRunningState state) throws SQLException { - // 利用 SQLRunningState.query() 通过真实连接查询数据 - String tableName = "t_config"; - DbTable table = state.getDbTable(tableName); - if (table != null) { - List> rows = state.query("SELECT * FROM t_config WHERE key = ?", params); - // 基于查询结果做处理 - } -} -``` - -## 注意事项 - -- **执行顺序**: 系统监听器（Insert/Update/Delete）order=100。自定义监听器 order < 100 时在 before 中先于系统监听器执行；order > 100 时则在系统监听器之后 -- **系统监听器不可移除**: `cleanCustomListeners()` 仅清除自定义监听器，Insert/Update/Delete 三个系统监听器始终存在 -- **before 抛异常阻止执行**: 在 `before()` 中抛出 `SQLException` 会中止当前 SQL 执行链，后续监听器的 before/after 均不会执行 -- **批量模式**: 当使用 JDBC batch 操作时，`SQLRunningState.isJdbcBatchMode()` 返回 true，`getBatchSQLRunningStateList()` 返回每条 SQL 的独立上下文 -- **线程安全**: `SQLRunningContext` 的监听器列表使用 `CopyOnWriteArrayList`，支持运行时动态注册。但建议在应用启动阶段一次性完成注册 -- **query() 走真实连接**: `SQLRunningState.query()` 使用未被代理的真实 `Connection`，不会触发监听器递归调用 -- **注册去重**: `SQLRunningContext.addListener()` 会检查监听器是否已存在（基于 equals），重复注册不会生效 - -## 变更说明 - -- **2026-04-28**: 同步最新代码，新增 `SQLRunningState.getBatchSQLRunningStateList()`、`getStatement()`、`getConnectionProxy()`、`getJdbcQuery()`、`triggerDBMetaData(String)` 方法文档；新增 `SQLRunningContext.getListeners()`、`addSystemListeners()` 方法文档；Maven 坐标版本号更新为 1.0.18 diff --git a/docs/components/index.md b/docs/components/index.md deleted file mode 100644 index 674535d..0000000 --- a/docs/components/index.md +++ /dev/null @@ -1,11 +0,0 @@ -# 组件索引 - -> 本文档由 `/rebuild-components-index` 自动生成。每次 `/write-components` 完成后应自动重建。 -> 手动维护本文件会被下一次重建覆盖。 - -| 组件模块 | 组件名称 | 组件描述（应用场景） | 文档地址 | -|----------|----------|----------------------|----------| -| dbstream-driver | DBEventPusher | 数据变更事件推送接口，在事务提交时接收 DBEvent 列表并推送到外部系统（消息队列、日志、缓存同步等）。 | [dbstream-driver_DBEventPusher](./dbstream-driver_DBEventPusher.md) | -| dbstream-driver | DBEventSupporter | 表级事件过滤接口，用于控制哪些数据库表启用数据变更事件捕获，默认全部支持。 | [dbstream-driver_DBEventSupporter](./dbstream-driver_DBEventSupporter.md) | -| dbstream-driver | DBStreamContext | DBStream 的全局统一入口，用于注册 SQL 执行监听器、事件推送者、表级过滤器，以及管理数据库元数据缓存。 | [dbstream-driver_DBStreamContext](./dbstream-driver_DBStreamContext.md) | -| dbstream-driver | SQLExecuteListener | SQL 执行拦截监听接口，允许在 SQL 执行前后插入自定义逻辑，如日志记录、性能监控、审计追踪等。 | [dbstream-driver_SQLExecuteListener](./dbstream-driver_SQLExecuteListener.md) | diff --git a/docs/plan/819340-change-dynamic-table-metadata-refresh.md b/docs/plan/819340-change-dynamic-table-metadata-refresh.md deleted file mode 100644 index dd25c81..0000000 --- a/docs/plan/819340-change-dynamic-table-metadata-refresh.md +++ /dev/null @@ -1,96 +0,0 @@ -# 调整计划: 动态表元数据刷新机制 - -> 编码: 819340 | 日期: 2026-04-28 | 类型: 调整任务 | 来源: change | 基于: docs/changes/819340-dynamic-table-metadata-refresh.md - -## 调整目标 - -为 DBStream 提供主动刷新 DBMetaData 缓存的公共 API，支持按表名或全量刷新元数据，使运行时动态创建或修改的表能被业务方手动触发更新后纳入 CDC 事件捕获。 - -## 影响面（基于指令结果） - -- 涉及 Maven 模块: dbstream-driver（单模块工程，`mvn help:evaluate` 确认 artifactId=dbstream-driver） -- 直接依赖者: 无（单模块工程，无 reactor 内部依赖） -- 外部 API / 公共组件影响: 是 — 新增公共 API 方法到 `DBStreamContext` 和 `DBMetaContext`，现有 API 行为不变 - -## 分步策略 - -### Step 1: 在 DBScanner 中新增单表扫描方法 - -新增 `scanTable(String tableName)` 方法，扫描指定单张表的元数据（列、主键），返回 `DbTable`。复用现有的 `loadDbTableInfo` 逻辑。 - -### Step 2: 在 DBMetaData 中新增主动刷新方法 - -新增 `refreshTable(ConnectionProxy, String tableName)` 方法：扫描指定表元数据并更新到 tables 列表。若表已存在则替换，不存在则追加。同时更新该表的本地序列化缓存。 - -新增 `refreshAll(ConnectionProxy)` 方法：重新扫描全部表，替换整个 tables 列表。 - -### Step 3: 在 DBMetaContext 中暴露刷新 API - -新增 `refreshTable(String jdbcKey, ConnectionProxy, String tableName)` 和 `refreshAll(String jdbcKey, ConnectionProxy)` 方法，委托给 `DBMetaData` 执行。 - -### Step 4: 在 DBStreamContext 中暴露公共 API - -新增对外方法： -- `refreshTable(String jdbcKey, String tableName)` — 刷新指定表的元数据 -- `refreshAll(String jdbcKey)` — 全量刷新指定数据源的元数据 - -### Step 5: 编写测试用例 - -验证手动调用 `refreshTable` / `refreshAll` 后，新表能被 CDC 捕获，修改后的列结构被正确使用。 - -## 新增文件 - -| 文件路径 | 用途说明 | -|----------|----------| -| `src/test/java/com/example/dbstream/tests/MetadataRefreshTest.java` | 手动刷新元数据的集成测试 | - -## 修改文件 - -| 文件路径 | 修改内容 | -|----------|----------| -| `src/main/java/com/codingapi/dbstream/scanner/DBScanner.java` | 新增 `scanTable(String tableName)` 方法，扫描单表元数据 | -| `src/main/java/com/codingapi/dbstream/scanner/DBMetaData.java` | 新增 `refreshTable(ConnectionProxy, String)` 和 `refreshAll(ConnectionProxy)` 方法 | -| `src/main/java/com/codingapi/dbstream/scanner/DBMetaContext.java` | 新增 `refreshTable` / `refreshAll` 方法 | -| `src/main/java/com/codingapi/dbstream/DBStreamContext.java` | 新增 `refreshTable(String jdbcKey, String tableName)` 和 `refreshAll(String jdbcKey)` 公共 API | - -## 移除文件（若有） - -无。 - -## 兼容性与迁移 - -- 保留的 API / 行为: 所有现有公共 API（`clear`/`clearAll`/`addUpdateSubscribe`/`getTable` 等）行为不变 -- 破坏性变更: 无 -- 数据/配置迁移: 无。现有 `.dbstream/` 本地缓存格式不变，`refreshTable` 会自动更新缓存文件 - -## 核验机制 - -| 验证项 | 说明 | -|--------|------| -| 编译 | `./mvnw compile` | -| 单测 | `./mvnw test` — 包含新增的元数据刷新测试和全部既有测试 | -| 打包 | `./mvnw package` | -| 行为回归 | 现有 User1/User2/User3 RepositoryTest 全部通过；新增 MetadataRefreshTest 验证手动刷新场景 | - -具体验证命令： - -```bash -./mvnw compile -./mvnw test -./mvnw package -``` - -## 执行顺序 - -1. 修改 `src/main/java/com/codingapi/dbstream/scanner/DBScanner.java` — 新增 `scanTable(String)` 方法 -2. 修改 `src/main/java/com/codingapi/dbstream/scanner/DBMetaData.java` — 新增 `refreshTable()` 和 `refreshAll()` 方法 -3. 修改 `src/main/java/com/codingapi/dbstream/scanner/DBMetaContext.java` — 新增 `refreshTable()` 和 `refreshAll()` 方法 -4. 修改 `src/main/java/com/codingapi/dbstream/DBStreamContext.java` — 新增公共 API -5. 创建 `src/test/java/com/example/dbstream/tests/MetadataRefreshTest.java` — 集成测试 -6. 执行 `./mvnw test` 验证 - -## 变更说明 - -**调整意见原文**: "仅提供提供主动刷新 API（按表名或全量刷新）的能力即可" - -**调整原因**: 缩小实现范围，仅提供手动刷新 API。移除了原计划中的 DDL 自动检测相关步骤（原 Step 1 SQLUtils DDL 检测、原 Step 6 DDLMetadataRefreshListener、原 Step 7 注册 DDL 监听器），以及对应的测试文件 DDLMetadataRefreshTest 和对 SQLUtils / SQLRunningContext 的修改。DDL 自动检测可作为后续独立任务实现。 diff --git a/pom.xml b/pom.xml index 49b0da2..4a7f09b 100644 --- a/pom.xml +++ b/pom.xml @@ -4,7 +4,7 @@ 4.0.0 com.codingapi.dbstream dbstream-driver - 1.0.18 + 1.0.19 https://github.com/codingapi/dbstream-driver dbstream-driver