如何将两张Oracle表同步至Redis单一JSON结构且无需物化视图

完全可以实现，以下是几种无需在Oracle端创建物化视图的可行方案，覆盖不同实时性需求和架构复杂度：

方案1：Oracle触发器 + PL/SQL Redis客户端 #

通过在employee_info和employee_detail表上创建行级触发器，监听增、删、改操作，触发时拼接两张表的关联数据为JSON，直接调用Redis的API更新对应键值。

实现步骤： #

1. 部署Oracle端的Redis PL/SQL客户端（比如开源的redis-oracle包），或者用UTL_HTTP调用Redis的REST接口（如Redis Stack的REST接口）。
2. 为两张表分别创建AFTER INSERT/UPDATE/DELETE触发器：
   • 触发器触发时，根据当前操作的employee_id关联查询另一张表的最新数据。
   • 将两张表的字段拼接为符合需求的JSON结构（比如{"id": 123, "name": "张三", "department": "技术部", "address": "北京市..."}）。
   • 调用Redis的JSON.SET命令（或普通SET命令，如果存字符串JSON），以employee:{id}为键存储合并后的JSON。

示例代码片段（触发器核心逻辑）： #

CREATE OR REPLACE TRIGGER trg_employee_sync_redis
AFTER INSERT OR UPDATE OR DELETE ON employee_info
FOR EACH ROW
DECLARE
    v_employee_json CLOB;
    v_employee_detail employee_detail%ROWTYPE;
BEGIN
    -- 关联查询详情表数据
    IF NOT DELETING THEN
        SELECT * INTO v_employee_detail FROM employee_detail WHERE employee_id = :NEW.employee_id;
        -- 拼接JSON（可使用Oracle的JSON_OBJECT函数）
        v_employee_json := JSON_OBJECT(
            'id' VALUE :NEW.employee_id,
            'name' VALUE :NEW.emp_name,
            'gender' VALUE :NEW.gender,
            'department' VALUE v_employee_detail.department,
            'address' VALUE v_employee_detail.home_address
        ).TO_CLOB();
        -- 调用Redis客户端写入数据
        redis.set('employee:' || :NEW.employee_id, v_employee_json);
    ELSE
        -- 删除操作时移除Redis中的键
        redis.del('employee:' || :OLD.employee_id);
    END IF;
EXCEPTION
    WHEN NO_DATA_FOUND THEN
        -- 处理详情表无对应数据的情况
        NULL;
END;
/

优缺点： #

• ✅ 实时性强，数据变更后立即同步
• ✅ 无需额外中间组件
• ❌ 触发器会增加Oracle的事务开销，高并发场景下可能影响原表性能
• ❌ 触发器逻辑需要维护，若表结构变更需同步修改触发器

方案2：基于CDC（变更数据捕获）的无侵入同步 #

使用Debezium这类CDC工具，直接读取Oracle的Redo Log捕获数据变更，无需在Oracle端做任何修改（无需触发器、物化视图），然后通过数据处理环节合并两张表的数据，写入Redis。

实现步骤： #

1. 配置Debezium Oracle连接器，监听employee_info和employee_detail两张表的变更事件。
2. 利用Kafka Connect的转换规则（SMT）或自定义处理器，根据employee_id关联两张表的变更数据：
   • 当任意一张表发生变更时，拉取另一张表的最新数据（可通过Debezium的快照或关联查询）。
   • 将两张表的字段合并为单一JSON结构。
3. 配置Kafka Connect的Redis连接器，将合并后的JSON写入Redis，以employee:{id}为键存储。

核心逻辑说明： #

Debezium会捕获每张表的增量变更（包括新增、更新、删除），你可以通过自定义的Kafka Streams应用或者SMT转换，将同属一个employee_id的两张表数据合并成完整的JSON对象，再推送到Redis。比如当employee_detail更新时，先获取对应employee_info的最新数据，合并后更新Redis中的键。

优缺点： #

• ✅ 完全无侵入Oracle，不影响源库性能
• ✅ 支持实时同步，且能处理历史数据快照初始化
• ✅ 扩展性好，可对接其他下游系统
• ❌ 需要部署Debezium、Kafka等中间组件，架构复杂度较高

方案3：定时增量查询同步 #

适合对实时性要求不高（比如分钟级延迟）的场景，通过定时任务（Python/Java脚本、Oracle DBMS_SCHEDULER）定期查询两张表的增量数据，合并后更新Redis。

实现步骤： #

1. 确保两张表有更新时间戳字段（如last_updated）或自增主键，用于识别增量数据。
2. 编写定时任务：
   • 记录上次同步的最大时间戳（或主键值）。
   • 查询两张表中last_updated大于上次时间戳的记录，按employee_id分组关联。
   • 合并每组数据为JSON，批量写入Redis（使用MSET或JSON.MSET提高效率）。
   • 更新下次同步的时间戳。

示例Python代码片段： #

import cx_Oracle
import redis
import json

# 初始化Oracle和Redis连接
oracle_conn = cx_Oracle.connect("user/password@oracle_host:1521/service")
redis_conn = redis.Redis(host='redis_host', port=6379, db=0)

# 获取上次同步的时间戳（可存在Redis或配置文件中）
last_sync_time = redis_conn.get('last_employee_sync_time') or b'2000-01-01 00:00:00'
last_sync_time = last_sync_time.decode('utf-8')

# 查询增量数据并关联
cursor = oracle_conn.cursor()
query = """
    SELECT ei.employee_id, ei.emp_name, ei.gender, ed.department, ed.home_address, ei.last_updated
    FROM employee_info ei
    LEFT JOIN employee_detail ed ON ei.employee_id = ed.employee_id
    WHERE ei.last_updated > :last_time OR ed.last_updated > :last_time
"""
cursor.execute(query, last_time=last_sync_time)

# 批量写入Redis
pipe = redis_conn.pipeline()
max_sync_time = last_sync_time
for row in cursor:
    emp_id, name, gender, dept, address, update_time = row
    emp_json = json.dumps({
        "id": emp_id,
        "name": name,
        "gender": gender,
        "department": dept,
        "address": address
    })
    pipe.set(f"employee:{emp_id}", emp_json)
    if update_time.strftime('%Y-%m-%d %H:%M:%S') > max_sync_time:
        max_sync_time = update_time.strftime('%Y-%m-%d %H:%M:%S')

pipe.execute()
# 更新上次同步时间
redis_conn.set('last_employee_sync_time', max_sync_time)

# 关闭连接
cursor.close()
oracle_conn.close()
redis_conn.close()

优缺点： #

• ✅ 实现简单，无需复杂组件
• ✅ 对Oracle性能影响极小（仅定时查询）
• ❌ 存在延迟，无法做到实时同步
• ❌ 需要维护增量查询的逻辑，若表结构变更需同步修改查询语句

内容的提问来源于stack exchange，提问作者Dharmesh Porwal