添加一个新的复制表

源数据库

添加附加日志

ALTER TABLE "REP"."C" ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY, FOREIGN KEY, UNIQUE INDEX) COLUMNS;

实例化准备

DBMS_CAPTURE_ADM.PREPARE_TABLE_INSTANTIATION(

table_name => 'rep.c',

supplemental_logging => 'keys');

添加捕捉规则

dbms_streams_adm.add_table_rules(

    table_name => '"REP"."C"',

    streams_type => 'CAPTURE',

    streams_name => '"TEST$CAP"',

    queue_name => '"STREAMADMIN"."TEST$CAPQ"',

    include_dml => TRUE,

    include_ddl => TRUE,

    include_tagged_lcr => TRUE,

    source_database => 'TEST.REGRESS.RDBMS.DEV.US.ORACLE.COM',

    inclusion_rule => TRUE);

添加传送规则

dbms_streams_adm.add_table_propagation_rules(

    table_name => '"REP"."C"',

    streams_name => '',

    source_queue_name => '"STREAMADMIN"."TEST$CAPQ"',

    destination_queue_name => '"STREAMADMIN"."TEST$APPQ"@RMAN.REGRESS.RDBMS.DEV.US.ORACLE.COM',

    include_dml => TRUE,

    include_ddl => TRUE,

    include_tagged_lcr => TRUE,

    source_database => 'TEST.REGRESS.RDBMS.DEV.US.ORACLE.COM',

    inclusion_rule => TRUE,

    and_condition => NULL,

    queue_to_queue => TRUE);

目标数据库

导入对象并实例化

使用imp数据时加入参数STREAMS_INSTANTIATION=Y(生效)

或者得到源数据库的当前scn

select dbms_flashback.get_system_change_number from dual;

DBMS_APPLY_ADM.SET_TABLE_INSTANTIATION_SCN(

source_object_name => 'rep.c',

source_database_name => ' TEST.REGRESS.RDBMS.DEV.US.ORACLE.COM ',

instantiation_scn => 1234567);                               --scn号

添加应用规则

  dbms_streams_adm.add_table_rules(

    table_name => '"REP"."C"',

    streams_type => 'APPLY',

    streams_name => '',

    queue_name => '"STREAMADMIN"."TEST$APPQ"',

    include_dml => TRUE,

    include_ddl => TRUE,

    include_tagged_lcr => TRUE,

    source_database => 'TEST.REGRESS.RDBMS.DEV.US.ORACLE.COM',

inclusion_rule => TRUE);

其他

搭建表级流复制系统代码

启用/禁用捕捉进程

  dbms_capture_adm.start_capture(

capture_name => '"TEST$CAP"');

dbms_capture_adm.stop_capture(

capture_name => '"TEST$CAP"');

启用/禁用传输

  dbms_aqadm.enable_propagation_schedule(

    queue_name => '"STREAMADMIN"."RMAN$CAPQ"',

    destination => 'TEST.REGRESS.RDBMS.DEV.US.ORACLE.COM',

destination_queue => '"STREAMADMIN"."RMAN$APPQ"');

  dbms_aqadm.disable_propagation_schedule(

    queue_name => '"STREAMADMIN"."TEST$CAPQ"',

    destination => 'CHARGE.REGRESS.RDBMS.DEV.US.ORACLE.COM',

destination_queue => destn_q);

启用/禁用应用进程

  dbms_apply_adm.start_apply(

apply_name => apply_nm_dqt);

  dbms_apply_adm.stop_apply(

apply_name => apply_nm_dqt);

加入负规则集

（不传送本机应用标签的LCR）

  dbms_streams_adm.add_table_propagation_rules(

    table_name => '"REP"."A"',

    streams_name => '',

    source_queue_name => '"STREAMADMIN"."TEST$CAPQ"',

    destination_queue_name => '"STREAMADMIN"."TEST$APPQ"@RMAN.REGRESS.RDBMS.DEV.US.ORACLE.COM',

    include_dml => TRUE,

    include_ddl => FALSE,

    include_tagged_lcr => TRUE,

    source_database => 'TEST.REGRESS.RDBMS.DEV.US.ORACLE.COM',

   inclusion_rule => FALSE,

   and_condition =>  ':lcr.get_tag() = HEXTORAW ("00") ',       --00为本机应用进程的标签

    queue_to_queue => true);

加入update冲突解决

DECLARE

cols DBMS_UTILITY.NAME_ARRAY;

BEGIN

cols(1) := 'c1';                          --列名

cols(2) := 'c2';                               --列名

DBMS_APPLY_ADM.SET_UPDATE_CONFLICT_HANDLER(

object_name => 'hr.jobs',                      --表名

method_name => 'OVERWRITE',                    --overwrite,discard,maximum,minmum,null

resolution_column => 'job_title',              --

column_list => cols);                          --列集合

END;

/

自定义冲突解决

BEGIN

DBMS_APPLY_ADM.SET_DML_HANDLER(

object_name => 'hr.locations',                          --表名

object_type => 'TABLE',                                 --

operation_name => 'UPDATE',                             --升级冲突

error_handler => false,                                

user_procedure => 'strmadmin.history_dml',              --自定义存储过程 Null取消

apply_database_link => NULL,

apply_name => NULL);

END;

/

自定义存储过程

CREATE OR REPLACE PROCEDURE history_dml(in_any IN ANYDATA)

IS

lcr SYS.LCR$_ROW_RECORD;

rc PLS_INTEGER;

BEGIN

-- Access the LCR

rc := in_any.GETOBJECT(lcr);

-- Insert information about the LCR into the history_row_lcrs table

INSERT INTO strmadmin.history_row_lcrs VALUES

(SYSDATE, lcr.GET_SOURCE_DATABASE_NAME(), lcr.GET_COMMAND_TYPE(),

lcr.GET_OBJECT_OWNER(), lcr.GET_OBJECT_NAME(), lcr.GET_TAG(),

lcr.GET_TRANSACTION_ID(), lcr.GET_SCN(), lcr.GET_COMMIT_SCN,

lcr.GET_VALUES('old'), lcr.GET_VALUES('new', 'n'));

-- Apply row LCR

lcr.EXECUTE(true);

END;/