ITPub博客

首页 > Linux操作系统 > Linux操作系统 > 数据复制_Stream

数据复制_Stream

原创 Linux操作系统 作者:redhouser 时间:2011-06-01 09:36:33 0 删除 编辑

本例子实现DML的捕获,应用,队列,没有涉及传播,规则等。

操作系统版本:
Linux net95 2.6.18-164.el5xen #1 SMP Thu Sep 3 04:47:32 EDT 2009 i686 i686 i386 GNU/Linux

Oracle版本:
Oracle Database 10g Enterprise Edition Release 10.2.0.1.0 - Production
With the Partitioning, OLAP and Data Mining options


--**********************1.1,准备-数据源*******************
conn system/obss;

create tablespace stream_tbs
 datafile 'c:\stream_tbs_01.dbf' size 10m;

create user hr identified by hr default tablespace stream_tbs;

grant resource,connect to hr;

connect hr/hr;

CREATE TABLE employees(
  employee_id    NUMBER(6),
  first_name     VARCHAR2(20),
  department_id  NUMBER(4));

insert into employees values(1,'Mike',1);
insert into employees values(2,'John',1);
insert into employees values(3,'Jack',2);

commit;

CREATE UNIQUE INDEX employees_id_pk ON employees (employee_id);

ALTER TABLE employees ADD (CONSTRAINT employees_id_pk PRIMARY KEY (employee_id));

--*********************1.2,准备-目的地******************************
conn system/obss;

create tablespace stream_new_tbs
  datafile 'c:\stream_new_tbs_01.dbf' size 10m;

create user hr_new identified by hr_new default tablespace stream_new_tbs;

grant resource,connect to hr_new;

connect hr_new/hr_new

CREATE TABLE employees_all(
  employee_id    NUMBER(6),
  first_name     VARCHAR2(20),
  department_id  NUMBER(4),
  timestamp date);

CREATE UNIQUE INDEX employees_all_id_pk ON employees_all (employee_id);

ALTER TABLE employees_all ADD (CONSTRAINT employees_all_id_pk PRIMARY KEY (employee_id));

--*********************1.3,准备-捕获与应用者************************
conn system/obss;

create tablespace stream_queue_tbs
  datafile 'c:\stream_queue_tbs_01.dbf' size 10m;

GRANT DBA TO strmadmin IDENTIFIED BY strmadmin;

--ACCEPT streams_tbs PROMPT 'Enter Streams administrator tablespace on cpap.net: '

ALTER USER strmadmin DEFAULT TABLESPACE stream_queue_tbs
                     QUOTA UNLIMITED ON stream_queue_tbs;

--授予stream 管理权限
connect sys/obss as sysdba;
BEGIN
  DBMS_STREAMS_AUTH.GRANT_ADMIN_PRIVILEGE(
    grantee          => 'strmadmin',   
    grant_privileges => true);
END;
/

connect  system/obss;
GRANT ALL ON hr_new.employees_all TO STRMADMIN;

-- 建立队列(anydata queue):streams_queue,队列表:streams_queue_table
conn strmadmin/strmadmin;
BEGIN
  DBMS_STREAMS_ADM.SET_UP_QUEUE(
    queue_table  => 'strmadmin.streams_queue_table',
    queue_name   => 'strmadmin.streams_queue');
END;
/


--*************************2,捕获与应用***********************************
CONNECT strmadmin/strmadmin;

--配置捕获过程
BEGIN
  DBMS_STREAMS_ADM.ADD_TABLE_RULES(
    table_name     => 'hr.employees',  
    streams_type   => 'capture',
    streams_name   => 'capture_emp',
    queue_name     => 'strmadmin.streams_queue',
    include_dml    =>  true,
    include_ddl    =>  false,
    inclusion_rule =>  true);
END;
/

--设置 hr.employees实例化 scn(如果不在同一个数据库,需要实例化)
DECLARE
  iscn  NUMBER;         -- Variable to hold instantiation SCN value
BEGIN
  iscn := DBMS_FLASHBACK.GET_SYSTEM_CHANGE_NUMBER();
  DBMS_APPLY_ADM.SET_TABLE_INSTANTIATION_SCN(
    source_object_name    => 'hr.employees',
    source_database_name  => 'ora10',
    instantiation_scn     => iscn);
END;
/

--create dml handler
CREATE OR REPLACE PROCEDURE emp_dml_handler(in_any IN ANYDATA) IS
  lcr          SYS.LCR$_ROW_RECORD;
  rc           PLS_INTEGER;
  command      VARCHAR2(30);
  old_values   SYS.LCR$_ROW_LIST;
BEGIN   
  -- Access the LCR
  rc := in_any.GETOBJECT(lcr);
  -- Get the object command type
  command := lcr.GET_COMMAND_TYPE();
 
  --Set the object_owner
  lcr.SET_OBJECT_OWNER('HR_NEW');
  -- Set the object_name in the row LCR
  lcr.SET_OBJECT_NAME('EMPLOYEES_ALL');

  -- Check for DELETE command on the hr.employees table
  IF command = '+++DELETE' THEN
    -- Set the command_type in the row LCR to INSERT
    lcr.SET_COMMAND_TYPE('INSERT');
    -- Get the old values in the row LCR
    old_values := lcr.GET_VALUES('old');
    -- Set the old values in the row LCR to the new values in the row LCR
    lcr.SET_VALUES('new', old_values);
    -- Set the old values in the row LCR to NULL
    lcr.SET_VALUES('old', NULL);
    -- Add a SYSDATE value for the timestamp column
    lcr.ADD_COLUMN('new', 'TIMESTAMP', ANYDATA.ConvertDate(SYSDATE));
    -- Apply the row LCR as an INSERT into the hr.emp_del table
    lcr.EXECUTE(true);
  elsIF command = 'DELETE' THEN
    -- Set the command_type in the row LCR to INSERT
    lcr.SET_COMMAND_TYPE('UPDATE');
    -- Add a SYSDATE value for the timestamp column
    lcr.ADD_COLUMN('new', 'TIMESTAMP', ANYDATA.ConvertDate(SYSDATE));
    -- Apply the row LCR as an INSERT into the hr.emp_del table
    lcr.EXECUTE(true);
 -- elsIF command = 'DELETE' THEN
 --  update hr_new.employees_all set timestamp=sysdate where employee_id=(lcr.get_value('old','EMPLOYEE_ID') ).getNumber();
 --  commit;
  else    --insert,update
    lcr.execute(true);
  END IF;
END;
/

--set dml_handler  for source table:INSERT/UPDATE/DELETE
BEGIN
  DBMS_APPLY_ADM.SET_DML_HANDLER(
    object_name         => 'hr.employees',
    object_type         => 'TABLE',
    operation_name      => 'INSERT',
    error_handler       => false,
    user_procedure      => 'strmadmin.emp_dml_handler',
    apply_database_link => NULL,
    apply_name          => NULL);
END;
/

BEGIN
  DBMS_APPLY_ADM.SET_DML_HANDLER(
    object_name         => 'hr.employees',
    object_type         => 'TABLE',
    operation_name      => 'UPDATE',
    error_handler       => false,
    user_procedure      => 'strmadmin.emp_dml_handler',
    apply_database_link => NULL,
    apply_name          => NULL);
END;
/

BEGIN
  DBMS_APPLY_ADM.SET_DML_HANDLER(
    object_name         => 'hr.employees',
    object_type         => 'TABLE',
    operation_name      => 'DELETE',
    error_handler       => false,
    user_procedure      => 'strmadmin.emp_dml_handler',
    apply_database_link => NULL,
    apply_name          => NULL);
END;
/

--创建队列的消息消费者(在队列中放入消息前必须设置消费者)
BEGIN
  DBMS_STREAMS_ADM.ADD_TABLE_RULES(
    table_name     => 'hr.employees',  
    streams_type   => 'dequeue',
    streams_name   => 'hr',
    queue_name     => 'strmadmin.streams_queue',
    include_dml    =>  true,
    include_ddl    =>  false,
    inclusion_rule =>  true);
END;
/

--配置应用过程
DECLARE
          emp_rule_name_dml  VARCHAR2(30);
          emp_rule_name_ddl  VARCHAR2(30);
BEGIN
  DBMS_STREAMS_ADM.ADD_TABLE_RULES(
    table_name      => 'hr.employees',
    streams_type    => 'apply',
    streams_name    => 'apply_emp',
    queue_name      => 'strmadmin.streams_queue',
    include_dml     =>  true,
    include_ddl     =>  false,
    source_database => 'ora10',
    dml_rule_name   => emp_rule_name_dml,
    ddl_rule_name   => emp_rule_name_ddl);
  DBMS_APPLY_ADM.SET_ENQUEUE_DESTINATION(
    rule_name               =>  emp_rule_name_dml,
    destination_queue_name  =>  'strmadmin.streams_queue');
END;
/

--  创建消费消息的过程
CREATE OR REPLACE PROCEDURE emp_dq (consumer IN VARCHAR2) AS
  msg            ANYDATA;
  row_lcr        SYS.LCR$_ROW_RECORD;
  num_var        pls_integer;
  more_messages  BOOLEAN := true;
  navigation     VARCHAR2(30);
BEGIN
  navigation := 'FIRST MESSAGE';
  WHILE (more_messages) LOOP
    BEGIN
      DBMS_STREAMS_MESSAGING.DEQUEUE(
        queue_name   => 'strmadmin.streams_queue',
        streams_name => consumer,
        payload      => msg,
        navigation   => navigation,
        wait         => DBMS_STREAMS_MESSAGING.NO_WAIT);
      IF msg.GETTYPENAME() = 'SYS.LCR$_ROW_RECORD' THEN
        num_var := msg.GetObject(row_lcr);  
        DBMS_OUTPUT.PUT_LINE(row_lcr.GET_COMMAND_TYPE || ' row LCR dequeued');
      END IF;
      navigation := 'NEXT MESSAGE';
    COMMIT;
    EXCEPTION WHEN SYS.DBMS_STREAMS_MESSAGING.ENDOFCURTRANS THEN
                navigation := 'NEXT TRANSACTION';
              WHEN DBMS_STREAMS_MESSAGING.NOMOREMSGS THEN
                more_messages := false;
                DBMS_OUTPUT.PUT_LINE('No more messages.');
              WHEN OTHERS THEN
                RAISE;
    END;
  END LOOP;
END;
/

--启动应用过程
BEGIN
  DBMS_APPLY_ADM.SET_PARAMETER(
    apply_name  => 'apply_emp',
    parameter   => 'disable_on_error',
    value       => 'n');
END;
/
 
BEGIN
  DBMS_APPLY_ADM.START_APPLY(
    apply_name  => 'apply_emp');
END;
/

--启动捕获过程
BEGIN
  DBMS_CAPTURE_ADM.START_CAPTURE(
    capture_name  => 'capture_emp');
END;
/


--*************************3,验证*************************************
--检查无效对象
conn system/obss;
select * from dba_objects where owner in ('HR','HR_NEW','STRMADMIN') and status<>'VALID';

--验证规则上下文
CONNECT strmadmin/strmadmin;

SELECT RULE_OWNER, RULE_NAME FROM DBA_STREAMS_RULES
  WHERE STREAMS_NAME = 'APPLY_EMP' AND
        STREAMS_TYPE = 'APPLY' AND
        SCHEMA_NAME  = 'HR' AND
        OBJECT_NAME  = 'EMPLOYEES' AND
        RULE_TYPE    = 'DML'
  ORDER BY RULE_NAME;


COLUMN RULE_OWNER HEADING 'Rule Owner' FORMAT A15
COLUMN DESTINATION_QUEUE_NAME HEADING 'Destination Queue' FORMAT A30

SELECT RULE_OWNER, DESTINATION_QUEUE_NAME
  FROM DBA_APPLY_ENQUEUE
  WHERE RULE_NAME =any(SELECT RULE_NAME FROM DBA_STREAMS_RULES
  WHERE STREAMS_NAME = 'APPLY_EMP' AND
        STREAMS_TYPE = 'APPLY' AND
        SCHEMA_NAME  = 'HR' AND
        OBJECT_NAME  = 'EMPLOYEES' AND
        RULE_TYPE    = 'DML')
  ORDER BY DESTINATION_QUEUE_NAME;


--在源表进行dml操作
connect hr/hr;
insert into employees values(4,'Mike',1);
commit;

update employees set first_name='AAA';
commit;

delete from employees where employee_id=3;
commit;

--在目标表检查结果:
CONNECT strmadmin/strmadmin;

SELECT employee_id, first_name, timestamp
  FROM hr_new.employees_all ORDER BY employee_id;

SELECT MSG_ID, MSG_STATE, CONSUMER_NAME
  FROM AQ$STREAMS_QUEUE_TABLE ORDER BY MSG_ID;

--dequeue message:
SET SERVEROUTPUT ON SIZE 100000
EXEC emp_dq('HR');


--***************clear site***************
conn system/obss;

drop user hr cascade;
drop user hr_new cascade;
drop user strmadmin cascade;

drop tablespace stream_tbs including contents and datafiles;
drop tablespace stream_new_tbs including contents and datafiles;
drop tablespace stream_queue_tbs including contents and datafiles;

 


 

来自 “ ITPUB博客 ” ,链接:http://blog.itpub.net/18922393/viewspace-696838/,如需转载,请注明出处,否则将追究法律责任。

请登录后发表评论 登录
全部评论

注册时间:2011-05-26

  • 博文量
    211
  • 访问量
    810205