Oracle 使用DB_Link同步MySQL数据

mahan

一、创建DB_LINK
1、Oracle验证MySQL驱动
注意：Oracle数据库在安装过程中会自动安装MySQL的驱动“dg4odbc”。目前仅需要验证该驱动即可。当前环境中存在该驱动。

$ file $ORACLE_HOME/bin/dg4odbc



2、配置驱动连接
注意：此处需要在Oracle数据库中配置驱动连接文件。同时，MySQL数据库是严重区分大小写，需要注意驱动文件中的大小敏感。


$ vi $ORACLE_BASE/odbc.ini
[myodbc5]
   Driver = /home/dbs/app/myodbc-x.x.x/lib/libmyodbc5.so
   Description = Connector/ODBC x.x Driver DSN
   SERVER = <IP ADDRESS>
   PORT = <MYSQL PORT NUMBER>
   USER = mysql_user
   PASSWORD = *****
   DATABASE = test
   OPTION = 0
   TRACE = OFF
3、配置监听
注意：此处需要在listener.ora和tnsnames.ora文件中均需要配置MYSQL监听。

# tnsname.ora文件
$ vi $ORACLE_HOME/rdbms/admin/[SID]/tnsnames.ora
myodbc5 =
(DESCRIPTION=
(ADDRESS=
(PROTOCOL=TCP) (HOST=[MYSQL HOST]) (PORT=1521)
)
(CONNECT_DATA=
(SID=myodbc5))
(HS=OK)
)
# listener.ora文件
$ vi $ORACLE_HOME/network/admin/listener.ora
SID_LIST_LISTENER=
     (SID_LIST=
        (SID_DESC=
             (SID_NAME=myodbc5)
             (ORACLE_HOME=/home/dbs/app/Ora/product/11.2.0/dbhome_1)
             (PROGRAM=dg4odbc)
            (ENV="LD_LIBRARY_PATH=/home/dbs/app/unixodbc-2.2.14/lib:/home/dbs/app/Ora/product/11.2.0/dbhome_1/lib")
       )
   )
4、配置透明网关
注意：此处网关为数据库的，不是OS的网关

$ vi $ORACLE_HOME/hs/admin/initmyodbc5.ora
HS_FDS_CONNECT_INFO=myodbc5
# Data source name in odbc.ini
HS_FDS_TRACE_LEVEL=OFF  -- 关闭追踪，降低负载，如出现错误，可以开启追踪，日志文件在$ORACLE_HOME/hs/ 下
HS_FDS_SHAREABLE_NAME=/home/dbs/app/unixodbc-2.3/lib/libodbc.so  ß 系统中该文件的路径
HS_FDS_SUPPORT_STATISTICS=FALSE
HS_LANGUAGE= AMERICAN_ AMERICA. AL32UTF8
AMERICAN_AMERICA.WE8ISO8859P15ß Oracle建议当前值，但也可以输入Oracle字符集，原因是ODBC可能不识别字符集
#
# ODBC env variables
set ODBCINI=/home/dbs/etc/odbc.ini
5、重载监听
$ lsnrctl reload – 重新加载

6、创建DB_LINK
# 注意：myodbc5是tnsnames.ora文件中的监听服务名

SQL> create public database link OM_LINK connect to "XXXX" identified by "XXXXXX" using'myodbc5';
7、测试连接即可
SQL> select * from "tab1"@myodbc5;
二、新建同步测试案例
1、建立同步表（Oracle数据库）
注意：需要从MySQL同步数据过来，建立对应的字段，并且设置好数据类型即可。当前MySQL和Oracle的字符集分别是“utf8mb4”和“AMERICAN_ AMERICA. AL32UTF8”

SQL>create table test_t1(id integer,create_date date);
2、新建同步记录表
注意：因为数据是不断变化的，在做增量同步时，需要考虑一个关键值且该值时有规律的变化或者主键ID、时间戳都可以。当前经了解其ID是使用系统自增的，那么可以通过ID来判断哪些数据是新增的，将最后同步的最大的ID写到同步记录表中，下次自动同步时获取该值与MySQL数据库中的表进行比对，比该值小的及同步到Oracle数据库中。

create table om_sync_control (
    table_name       varchar2(50) primary key,    -- 被同步的表名
    last_sync_id     number(18),                 -- 上次同步的最大ID
    last_sync_time   date,                        -- 上次同步时间
    sync_status      varchar2(10)                -- 状态：SUCCESS/FAILED
);
3、创建错误信息表
注意：该表用来记录每次同步数据时是否发生报错，如果出现报错将记录对应的报错信息

CREATE TABLE om_sync_error_log (
    table_name  VARCHAR2(50),   -- 被同步的表名
    ids         NUMBER(20),    -- 失败的ID
    om_error    BLOB,  -- 错误原因
    insert_date DATE   -- 插入日期
);
4、初始化同步记录表
SQL> insert into om_sync_control (table_name, last_sync_id, last_sync_time, sync_status) values ('test_t1', 0, sysdate, 'SUCCESS');


5、创建存储过程
-- 创建同步数据存储过程

create or replace procedure SYNC_ON_INCREMENTAL is   
    v_max_id         number := 0;  -- 记录最大值
    v_count          number := 0;  -- 插入数据总量
    v_last_sync_id   number;       -- 上次同步的最大值
    v_sync_time      date := sysdate;  -- 同步时间
    -- 自定义异常
    e_sync_error exception
     -- 游标：查询上次同步后新增的数据
    cursor new_data_cur is
        select id, create_date
        from test_t1@OM_LINK
        where id > (select nvl(last_sync_id, 0)
                    from om_sync_control
                    where table_name = 'test_t1')
        order by id;  -- 按ID排序，确保顺序处理
begin
    -- 获取当前已同步的最大ID（用于后续更新）
    SELECT
        nvl(last_sync_id, 0)
    INTO v_last_sync_id
    FROM
        om_sync_control
    WHERE
        table_name = 'test_t1'
    FOR UPDATE;  -- 加锁，防止并发同步
    dbms_output.put_line('开始增量同步，上次同步ID：' || v_last_sync_id);

    -- 遍历新增数据
    for rec in new_data_cur loop
        begin
            -- 插入到Oracle目标表
            insert into test_t2 (
                id,            
                create_date
            ) values (
                rec.id,           -- 使用MySQL的自增ID
                rec.create_date
            );
            -- 记录本次循环的最大ID
            if rec.id > v_max_id then
                v_max_id := rec.id;
            end if;
            v_count := v_count + 1;
            -- 每100条提交一次，避免大事务
            if mod(v_count, 100) = 0 then
                commit;
                dbms_output.put_line('已同步 ' || v_count || ' 条，当前最大ID：' || v_max_id);
            end if;
        exception
            when others then
                -- 记录错误信息，继续处理下一条
                dbms_output.put_line('插入失败，ID=' || rec.id || '，错误：' || sqlerrm);
               --insert into om_sync_error_log values('test_t1',rec.id,'error',sysdate);
　　　　　　　　insert into om_sync_error_log values('test_t1',rec.id,'sqlerrm',sysdate);
               commit;
        end;
    end loop;

    -- 更新同步记录表
    if v_max_id > 0 then
        update om_sync_control
        set last_sync_id = v_max_id,
            last_sync_time = v_sync_time,
            sync_status = 'SUCCESS'
        where table_name = 'test_t1';
        commit;
        dbms_output.put_line('增量同步完成。共同步 ' || v_count || ' 条新数据，最大ID：' || v_max_id);
    else
        dbms_output.put_line('没有新数据需要同步。');
    end if;
exception
    when others then
        rollback;
        dbms_output.put_line('同步过程出错：' || sqlerrm);
        -- 更新状态为失败
        update om_sync_control
        set sync_status = 'FAILED',
            last_sync_time = sysdate
        where table_name = 'test_t1';
        commit;
        raise;
end SYNC_ON_INCREMENTAL;
6、配置定时任务
-- 开启定时任务
BEGIN
  dbms_scheduler.create_job(
   job_name=>'SYNC_ON_INCREMENTAL_FIRST',  --定时任务名称
   job_type => 'STORED_PROCEDURE',--类型为存储过程
   job_action=>'SYNC_ON_INCREMENTAL',--存储过程名称
   start_date => sysdate, --首次执行时间
   repeat_interval=>'FREQ=HOURLY;INTERVAL=12', --十二小时执行一次
   --【FREQ=minutely; 表示间隔单位，可选值Yearly,monthly,weekly,datly,hourly,minutely,secondly】
   enabled=>true  -- 立即启用定时任务
);
END;
7、查看定时任务是否启动
ENABLE：TRUE OR FALSE（标志定时任务是否启动）
STATE：RUNNING（标志定时任务运行中）、SCHEDULED（定时中）
-- 查询定时任务是否启动
Select * from dba_scheduler_jobs where job_name ='SYNC_ON_INCREMENTAL_FIRST' order by start_date desc;
8、查看定时任务具体执行信息
STATUS：SUCCEEDED和FAILED（定时任务执行成功或失败）
REQ_START_DTAE：定时任务执行时间
ERRORS：记录系统报错信息，如：违反唯一约束条件、程序包或函数无效等
OUTPUT：输出存储过程里面的结束时定义的提示语
select * from dba_scheduler_job_run_details where job_name ='SYNC_ON_INCREMENTAL_FIRST' order by log_date desc;
9、关闭定时任务

begin
dbms_scheduler.drop_job('SYNC_ON_INCREMENTAL_FIRST',force => true);  -- 删除定时任务，force：无论当前定时任务什么状态，都删除
  --dbms_scheduler.run_job('TEST_JOB4',true);  -- 手动运行定时任务
--dbms_scheduler.enable('TEST02');  -- 启用定时任务，根据定义的执行频率，Oracle会自动决定什么时候开始运行
--dbms_scheduler.disable('TEST_JOB2');
end;