stream 的delete、insert冲突解决方式

2021-12-19 06:56:35

3.继续测试：

SQL> create table hz.t6 (id number primary key ,vname varchar2(20),vaddr varchar2(20),vqty number);

Table created

SQL> insert into hz.t6 values (1,'a1','guangzhou',100);

1 row inserted

SQL> insert into hz.t6 values (2,'b1','zhenzhou',300);

1 row inserted

SQL> insert into hz.t6 values (3,'c1','zhenming',200);

1 row inserted

SQL> commit;

Commit complete

SQL> select * from hz.t6;

        ID VNAME                VADDR                      VQTY
---------- -------------------- -------------------- ----------
         1 a1                   guangzhou                   100
         2 b1                   zhenzhou                    300
         3 c1                   zhenming                    200

查看目标库，也是3条记录：

SQL> select * from hz.t6;

        ID VNAME                VADDR                      VQTY
---------- -------------------- -------------------- ----------
         1 a1                   guangzhou                   100
         2 b1                   zhenzhou                    300
         3 c1                   zhenming                    200

目标库做删除操作：
SQL> DELETE HZ.T6 WHERE ID=3;

1 row deleted

SQL> commit;

Commit complete

SQL> select * from hz.t6;

        ID VNAME                VADDR                      VQTY
---------- -------------------- -------------------- ----------
         1 a1                   guangzhou                   100
         2 b1                   zhenzhou                    300

原库也删除相同的序号3记录，
SQL> delete hz.t6 where id=3;

1 row deleted

SQL> commit;

Commit complete

目标库的应用进程显示abort.

SQL> select status from dba_apply;

STATUS
--------
ABORTED

应用进程的错误信息，
SQL> select LOCAL_TRANSACTION_ID,ERROR_MESSAGE    from dba_apply_error;

LOCAL_TRANSACTION_ID   ERROR_MESSAGE
---------------------- ----------------------------------------------------------------------
1.16.629               ORA-26787: The row with key ("ID") = (3) does not exist in table HZ.T1
                       ORA-01403: no data found

冲突解决
-------------------
冲突有很多类型，stream提供更新操作冲突类型的handler，其他类型的handler需要自己定义：

自定义handler的方法：
You use the SET_DML_HANDLER procedure in the DBMS_APPLY_ADM package to designate one or more custom conflict handlers for a particular table

CREATE TABLE strmadmin.history_row_lcrs(
timestamp DATE,
source_database_name VARCHAR2(128),
command_type VARCHAR2(30),
object_owner VARCHAR2(32),
object_name VARCHAR2(32),
tag RAW(10),
transaction_id VARCHAR2(10),
scn NUMBER,
commit_scn NUMBER,
old_values SYS.LCR$_ROW_LIST,
new_values SYS.LCR$_ROW_LIST)
NESTED TABLE old_values STORE AS old_values_ntab
NESTED TABLE new_values STORE AS new_values_ntab;

SQL> CREATE TABLE event_log (id NUMBER, timestamp DATE, event VARCHAR2(2000));

Table created

SQL> create sequence seq_event increment by 1 start with 1 nomaxvalue nocache;

Sequence created

CREATE OR REPLACE PROCEDURE hz_t5_handler(in_any IN SYS.ANYDATA) IS

lcr              SYS.LCR$_ROW_RECORD;
rc               PLS_INTEGER;
object_owner     VARCHAR2(30);
object_name      VARCHAR2(40);
dmlcommand       VARCHAR2(10);
v_name           varchar2(20) := ' ';
v_old_id_anydata SYS.ANYDATA;
v_old_id         number;
v_dummy          PLS_INTEGER;
v_sqlcode        varchar2(32);
v_sqlerrm        varchar2(255);
--v_typecode       PLS_INTEGER;
--v_type           SYS.ANYTYPE;
non_null_anytype_for_NUMBER exception;
unknown_typename exception;
v_cnt number;
BEGIN
-- Access the LCR
rc           := in_any.GETOBJECT(lcr);
object_owner := lcr.GET_OBJECT_OWNER();
object_name := lcr.GET_OBJECT_NAME();
dmlcommand   := lcr.GET_COMMAND_TYPE();
if dmlcommand in ('INSERT') then

    v_old_id_anydata := lcr.get_value('NEW', 'NO', 'Y');
elsif dmlcommand in ('DELETE') then
    v_old_id_anydata := lcr.get_value('OLD', 'NO');
else
    null;
end if;
-- 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'));
commit;

v_dummy := v_old_id_anydata.GetNUMBER(v_old_id /* OUT */);

DBMS_OUTPUT.PUT_LINE(TO_CHAR(rc) || To_Char(v_dummy)); --临时加的字段
select count(*) into v_cnt from hz.t6@lirhz.net a where a.id = v_old_id;
if v_cnt = 1 then
    select vname into v_name from hz.T6@lirhz.net a where a.id = v_old_id;

    IF object_owner = 'HZ' and object_name = 'T6' and dmlcommand IN ('INSERT' /*, 'DELETE', 'UPDATE'*/
       ) THEN

      -- Add Columns
      lcr.add_column('NEW', 'NAME', sys.anydata.convertvarchar2(v_name));
    ELSE
      lcr.add_column('OLD', 'NAME', sys.anydata.convertvarchar2(v_name));
    END IF;
    LCR.EXECUTE(TRUE);
elsif v_cnt = 0 then
    IF object_owner = 'HZ' and object_name = 'T6' and dmlcommand IN ('INSERT' /*, 'DELETE', 'UPDATE'*/
       ) THEN

      -- Add Columns
      lcr.add_column('NEW', 'NAME', sys.anydata.convertvarchar2(' '));
    ELSE
      lcr.add_column('OLD', 'NAME', sys.anydata.convertvarchar2(' '));
    END IF;
    LCR.EXECUTE(TRUE);
else
    null;
end if;

insert into event_log
    (id, timestamp, event)
values
    (seq_event.nextval,
     sysdate,
     'succeed! v_id=' || v_old_id || 'cnt=' || v_cnt || ',dmlcommand=' ||
     dmlcommand || ',v_name=' || v_name);
commit;
exception
when others then
    rollback;
    v_sqlcode := sqlcode;
    v_sqlerrm := sqlerrm;
    insert into event_log
      (id, timestamp, event)
    values
      (seq_event.nextval,
       sysdate,
       v_sqlcode || '|' || v_sqlerrm || ',v_old_id=' || v_old_id || ',cnt=' ||
       --v_cnt || ',v_name=' || v_name || ',dmlcommand=' || dmlcommand);
       v_cnt || ',v_name=' || v_name || ',dmlcommand=' || dmlcommand || TO_CHAR(rc)|| To_Char(v_dummy) );
    commit;

END;

SQL> BEGIN
2     DBMS_APPLY_ADM.SET_DML_HANDLER(
3     object_name => 'hz.t6',
4     object_type => 'TABLE',
5     operation_name => 'DELETE',
6     error_handler => false,
7     user_procedure => 'strmadmin.hz_t5_handler',
8     apply_database_link => NULL,
9     apply_name => NULL);
10    END;
11 /

PL/SQL procedure successfully completed

SQL> BEGIN
2     DBMS_APPLY_ADM.SET_DML_HANDLER(
3     object_name => 'hz.t6',
4     object_type => 'TABLE',
5     operation_name => 'INSERT',
6     error_handler => false,
7     user_procedure => 'strmadmin.hz_t5_handler',
8     apply_database_link => NULL,
9     apply_name => NULL);
10    END;
11 /

PL/SQL procedure successfully completed

删除之前的应用进程错误号信息，重新时应用日志应用。

SQL> exec dbms_apply_adm.execute_all_errors('APPLY_STANDY');

PL/SQL procedure successfully completed

SQL> exec dbms_apply_adm.start_apply('APPLY_STANDY');

PL/SQL procedure successfully completed

原库重新插入id=3的记录：
SQL> insert into hz.t6 values (3,'c1','zhenming',200);

1 row inserted

SQL> commit;

Commit complete

SQL> select * from hz.t6;

        ID VNAME                VADDR                      VQTY
---------- -------------------- -------------------- ----------
         1 a1                   guangzhou                   100
         2 b1                   zhenzhou                    300
         3 c1                   zhenming                    200

查看目标库，也是3条记录：

原库也删除相同的序号3记录，
SQL> delete hz.t6 where id=3;

1 row deleted

SQL> commit;

Commit complete

目标库的应用进程还是正常的.
SQL> select APPLY_NAME,status from dba_apply;

APPLY_NAME                     STATUS
------------------------------ --------
APPLY_STANDY                   ENABLED

SQL> select * from event_log;

        ID TIMESTAMP   EVENT
---------- ----------- -----------------------------------------------------------
         2 2010-10-15 -30625|ORA-30625: method dispatch on NULL SELF argument is disallowed,v_old_id=,cnt=,v_name= ,dmlcommand=DELETE

目标库插入需要6的记录：

SQL> insert into hz.t6 values (6,'e1','haikou',600);

1 row inserted

SQL> commit;

Commit complete

原库也插入目标为6的序号：
SQL> insert into hz.t6 values (6,'e1','haikou',600);

1 row inserted

SQL> commit;

Commit complete

SQL> select APPLY_NAME,status from dba_apply;

APPLY_NAME STATUS
------------------------------ --------
APPLY_STANDY ENABLED

SQL> select * from event_log;

        ID TIMESTAMP   EVENT
---------- ----------- --------------------------------------------------------------------------
         2 2010-10-15 -30625|ORA-30625: method dispatch on NULL SELF argument is disallowed,v_old_id=,cnt=,v_name= ,dmlcommand=DELETE
         3 2010-10-15 -30625|ORA-30625: method dispatch on NULL SELF argument is disallowed,v_old_id=,cnt=,v_name= ,dmlcommand=INSERT

至此t6表的插入,删除操作都已经正常进行，不会因为冲突而中断应用进程的应用了。

本文转自 gjm008 51CTO博客，原文链接：http://blog.51cto.com/gaoshan/406547，如需转载请自行联系原作者

码农公寓

相关文章