基于 Oracle LogMiner 的 CDC 日志解析

Yunjie Ge

发布于 2025-04-27 11:03:33

39900

代码可运行

文章被收录于专栏：数据库与编程数据库与编程

运行总次数：0

代码可运行

Oracle 的 Change Data Capture (CDC) 机制利用 LogMiner 解析重做日志获取数据变更。

CDC日志解析的核心在于精确重放事务：LogMiner 根据 SCN 顺序读取重做日志，将同一事务的所有 DML 先缓存，直到遇到 COMMIT 时才发往下游。这样可以保证即使多表间存在外键关系，也能按生产环境提交顺序执行，避免目标库出现不一致。对于 DDL 变更，启用离线字典时可使用 DDL_DICT_TRACKING 自动维护表结构版本，确保后续 DML 能正确解析。

在日志异常方面，通过 SKIP_CORRUPTION 选项让 LogMiner 跳过损坏块继续读取。若遇到缺失归档日志，代码会捕获错误并主动查询 V$LOGMNR_LOGS 找到漏掉的日志文件编号，再手动补入，完成断点续传。同时，引入commitmetascn、currentScn 等位点信息存储机制，以支持作业从任意中断点重启，保证 CDC 服务具备容错能力。

通过配置灵活的启动选项、高度依赖 Oracle 内部视图（如 VARCHIVED_LOG、VLOGMNR_CONTENTS）以及细致的事务管理逻辑，实现对 Oracle 日志的 CDC 解析，确保数据完整性和事务一致性。在实际使用时，还需注意开启归档模式和必要的补充日志，才能获取完整的变更记录。

Oracle LogMiner 的 CDC 日志解析整个过程主要包括：

连接数据库：建立 Oracle 连接，从上次位置获取起始 SCN 或时间戳；
确定字典模式：根据配置选择 Online（在线字典）、Offline（离线字典）或 External（外部字典）模式；
准备字典：
- Offline 模式：使用 DBMS_LOGMNR_D.BUILD 将数据字典导出到重做日志，并通过 V$ARCHIVED_LOG 确定包含字典的归档日志文件（dictionary_begin='YES'）并加入到 LogMiner；
- External 模式：加载指定的外部字典文件（如 ORACLE 导出的 .dct 文件），并按配置添加需要解析的归档日志文件；
加载重做日志：调用 DBMS_LOGMNR.ADD_LOGFILE 添加需要解析的归档日志列表；
启动 LogMiner 会话：调用 DBMS_LOGMNR.START_LOGMNR 并设置参数（起始 SCN/时间戳范围、DDL 跟踪、跳过损坏块等选项），启动增量日志解析；
查询重做内容：构造对 V$LOGMNR_CONTENTS 视图的 SELECT 查询，过滤所需的表和事务边界事件（START、COMMIT、ROLLBACK等）；
遍历结果集：按 SCN 顺序循环读取 V$LOGMNR_CONTENTS 返回的行，根据 OPERATION 类型分情况处理：
- START：启动新事务并缓存；
- DML（INSERT/UPDATE/DELETE/LOB_WRITE）：将变更组织到对应事务中，生成 DMLRecord 并处理特殊列（如 LOB）等；
- DDL：离线字典模式下利用 DDLRecord 解析并处理 DDL 变更；
- COMMIT/ROLLBACK：完成事务，若事务已包含变更则打包并发送到下游队列；
- 错误/缺失：遇到标记 CORRUPTED_BLOCKS（状态1343）或 DICTIONARY_DIFMISMATCH（状态2）时，根据配置跳过或终止，并记录警告。
结束与清理：关闭 ResultSet，根据需要调用 DBMS_LOGMNR.END_LOGMNR，并记录线程退出。

在以上步骤中，需要重点保证事务边界的正确性和 SCN 顺序，主要步骤：连接 Oracle -> 设置字典模式 -> 导出/加载字典 -> 添加重做日志 -> 启动 LogMiner -> 执行查询 -> 逐条处理日志记录 -> 组装并输出 CDC 变更 -> 结束解析。

开发实现时主要包含以下几个方法：启动会话方法（startSession）、转储字典到重做日志(dumpDictionaryToRedoLogs)、加载重做日志文件（loadRedoLogFiles）、添加包含字典快照的归档日志（addDictionaryLogFiles)、主循环解析（run）等方法。

一、startSession()方法

该方法负责准备并启动 LogMiner 会话。主要流程如下：

1. 参数初始化：读取先前保存的位点（SCN 或时间戳），确定起始解析点。若用户指定了起始时间，则先将时间转换为 SCN（通过 SELECT TIMESTAMP_TO_SCN(<ts>) 实现）。

2. 设置 LogMiner 选项：根据字典模式构造 DBMS_LOGMNR.START_LOGMNR 调用中的 OPTIONS。常用选项包括：

SKIP_CORRUPTION：跳过损坏块继续解析；
NO_ROWID_IN_STMT：在生成的 SQL 中不包含 ROWID（依赖补充日志保证唯一行标识）；
DICT_FROM_ONLINE_CATALOG/DICT_FROM_REDO_LOGS：选择从在线字典或重做日志中加载数据字典；
CONTINUOUS_MINE：持续监控日志，自动加入新日志（已弃用，仅为兼容性）；

DDL_DICT_TRACKING：让 LogMiner 跟踪 DDL 事件并更新内部字典（适用于离线字典模式）。例如，在线模式下会使用 DICT_FROM_ONLINE_CATALOG + CONTINUOUS_MINE；离线模式下使用 DICT_FROM_REDO_LOGS + CONTINUOUS_MINE + DDL_DICT_TRACKING。

3. 调用 START_LOGMNR：执行类似下面的语句：

BEGIN
  DBMS_LOGMNR.START_LOGMNR(
    OPTIONS => DBMS_LOGMNR.SKIP_CORRUPTION + …,
    STARTSCN => <起始SCN>,
    ENDSCN   => <结束SCN>… 
  );
END;

这将启动 LogMiner 会话并加载相应的字典和日志文件列表。

4. 构造查询语句：根据模式和监控表列表拼接对 V$LOGMNR_CONTENTS 的查询。注意：如果支持 PDB（容器数据库），查询中需要带上容器名；对 Offline 模式则专门包含 DDL 操作筛选。

5. 处理启动异常：如果 START_LOGMNR 失败并报错 1291/1292（通常表示缺失日志文件），需查询 VLOGMNR_LOGS 或 VARCHIVED_LOG 找出缺失的日志并手动添加。代码中对此会捕获异常并查询 V

二、dumpDictionaryToRedoLogs()方法：此方法用于 Offline 模式下导出数据字典到重做日志。它通过执行 DBMS_LOGMNR_D.BUILD(DictFilename=>null, DictSchema=>null, Options=>DBMS_LOGMNR_D.STORE_IN_REDO_LOGS) 将当前数据库的字典结构写入归档日志。之后，LogMiner 可在这些日志中查询到数据字典信息。

三、loadRedoLogFiles()方法：该方法根据配置的日志文件路径列表，将指定的归档日志逐个加到 LogMiner 会话中。它解析属性 RedologFiles（逗号分隔文件列表）并执行多次 DBMS_LOGMNR.ADD_LOGFILE(FILENAME=>..., OPTIONS=>DBMS_LOGMNR.NEW)。

四、addDictionaryLogFiles()：在离线字典模式下，需要将包含字典快照的归档日志添加到 LogMiner。该方法：

通过 V$ARCHIVED_LOG 找到最近包含字典起始 (DICTIONARY_BEGIN='YES') 和结束 (DICTIONARY_END='YES') 标记的日志序列；
确定对应的线程号和序列范围；
再次查询获得该线程对应的开始和结束序列号；
最后循环执行 ADD_LOGFILE 加入从起始序列到结束序列的所有日志文件。这样可以确保 LogMiner 能加载用于构建字典的所有重做日志片段。Oracle 文档指出，当使用 DICT_FROM_REDO_LOGS 时 LogMiner 会期望在指定的重做日志文件中找到字典。

五、run()：主循环方法，从 LogMiner 中读取并处理日志行，逻辑较复杂，包括：

准备查询：执行上一步构建的 SELECT ... FROM V$LOGMNR_CONTENTS，并设置合适的 fetchSize。
遍历结果集：按顺序处理每一行日志记录，根据 OPERATION 和 STATUS 字段区分：
- 状态字段检查：STATUS=1234/1343 表示遇到损坏块，此时记录跳过并打印警告；STATUS=2 表示字典版本不匹配，此处代码仅在 Online 模式下警告并继续。
- 转换操作类型：根据 OPERATION_CODE 得到枚举类型（如 START、INSERT、UPDATE 等）。
- START：代表新事务开始，创建 Transaction 对象并缓存（带有初始 SCN、会话信息等）。设置事务的逻辑日志位点 (LSN) 。
- DML (INSERT/UPDATE/DELETE/LOB_WRITE)：先通过 (SEG_OWNER, TABLE_NAME) 等信息匹配配置的监控表，如果未配置则跳过。然后获取对应事务，如果不存在且允许补齐已提交事务，则记录警告。构造 DMLRecord，合并跨行的 CSF (跨段 SQL) 继续读取完整的 SQL_REDO。处理过程中，针对 LOB 或自增列等特殊类型调用 SpecialColumnTypeHandler 进行转换。最终调用 processDMLRecord(sql, opType) 方法生成 CDC 变更记录放入输出队列。
- DDL：仅在离线模式下处理。利用 DDLRecord 对象解析 SQL_REDO 中的 DDL 语句，更新内部表结构版本。对于 DDL 记录所在的事务，也同样会在 commit 时发送这些变更。
- COMMIT：结束一个事务。查找事务缓存，如果事务中有 DML 记录则先将其刷新到队列，然后用本条记录更新事务的 commitScn、时间戳等元信息。调用 transaction.process(COMMIT) 触发事务发送逻辑，并从缓存中移除该事务。
- ROLLBACK：类似于 COMMIT，但标记撤销。若事务被处理过，也会将之前的 DML 添加到队列；然后清除缓存中的事务对象。
错误处理：在解析过程中，如果遇到异常（比如无法解析某条 DML），记录 ErrorRecord 并根据严重性决定是否中断。循环结束后关闭结果集，并根据线程状态决定是否调用 close() 清理 LogMiner 会话。

在这个过程中，每个事务的变更按照提交顺序发送下游，保证了端到端的事务一致性。在 run 结束后，如果没有正常调用 stop()，会产生致命错误记录以示警。

其中关键 SQL包括：

启动 LogMiner：调用 DBMS_LOGMNR.START_LOGMNR 时常用的 SQL 模板如：

BEGIN
  DBMS_LOGMNR.START_LOGMNR(
    OPTIONS => DBMS_LOGMNR.SKIP_CORRUPTION + DBMS_LOGMNR.NO_ROWID_IN_STMT
               + DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG + DBMS_LOGMNR.CONTINUOUS_MINE,
    STARTSCN => :start_scn,
    ENDSCN   => :end_scn
  );
END;

其中 DICT_FROM_ONLINE_CATALOG 表示使用在线字典，DICT_FROM_REDO_LOGS 表示使用重做日志中的字典；CONTINUOUS_MINE 让 LogMiner 自动加载后续日志；DDL_DICT_TRACKING 则开启 DDL 跟踪。

查询归档日志：在导出字典和加载日志时，会用到类似下面的查询：

-- 找出包含字典起点的日志
SELECT THREAD#, SEQUENCE#, FIRST_CHANGE# 
  FROM V$ARCHIVED_LOG
 WHERE DICTIONARY_BEGIN='YES' AND STATUS='A';
-- 找出当前 SCN 所在日志
SELECT THREAD#, SEQUENCE# 
  FROM V$ARCHIVED_LOG
 WHERE FIRST_CHANGE# <= :scn AND NEXT_CHANGE# > :scn
   AND ARCHIVED='YES' AND STATUS='A';

这里 FIRST_CHANGE#/NEXT_CHANGE# 范围用于定位包含特定 SCN 的日志文件。

时间戳转 SCN：为了按时间点启动解析，代码执行：

SELECT TIMESTAMP_TO_SCN(TO_TIMESTAMP(:start_time,'YYYY-MM-DD HH24:MI:SS')) 
  FROM DUAL;

TIMESTAMP_TO_SCN 返回某一时间点对应的系统 SCN，保证从正确的日志位置开始。

读取 LogMiner 内容：核心查询对 V$LOGMNR_CONTENTS，示例结构如下：

SELECT SCN, SQL_REDO, OPERATION_CODE, TIMESTAMP, XID, CSF, SEG_OWNER, TABLE_NAME, 
       OPERATION, ROW_ID, ROLLBACK, RS_ID
  FROM V$LOGMNR_CONTENTS
 WHERE SCN > :start_scn AND SCN <= :end_scn
   AND (
      (SEG_OWNER IN (…) AND TABLE_NAME IN (…))
      OR (OPERATION IN ('START','COMMIT','ROLLBACK'))
   );