Oracle 流stream将删除的数据保存-编程知识

Oracle 流stream将删除的数据保存

--实验的目的是捕获hr.employees表的删除行，将删除行插入到emp_del表中。
--设置初始化参数

AQ_TM_PROCESSES=1
COMPATIBLE=9.2.0
LOG_PARALLELISM=1

--查看数据库的名称，我的为ora9,将以下的ora9全部替换为你的数据库名称
--数据库为归档模式
--建立表emp_del,用于存放EMPLOYEES的删除数据

conn hr/hr
CREATE TABLE emp_del( employee_id    NUMBER(6), first_name     VARCHAR2(20), last_name      VARCHAR2(25), email          VARCHAR2(25), phone_number   VARCHAR2(20), hire_date      DATE, job_id         VARCHAR2(10), salary         NUMBER(8,2), commission_pct NUMBER(2,2), manager_id     NUMBER(6), department_id  NUMBER(4),timestamp      DATE);CREATE UNIQUE INDEX emp_del_id_pk ON emp_del (employee_id);ALTER TABLE emp_del ADD (CONSTRAINT emp_del_id_pk PRIMARY KEY (employee_id));

--建立管理用户,设定默认表空间，授权

conn / as sysdba
drop user strmadmin cascade;
GRANT CONNECT, RESOURCE, SELECT_CATALOG_ROLE TO strmadmin IDENTIFIED BY strmadmin;
ALTER USER strmadmin DEFAULT TABLESPACE users;GRANT ALL ON hr.emp_del  TO strmadmin;GRANT EXECUTE ON DBMS_APPLY_ADM        TO strmadmin;
GRANT EXECUTE ON DBMS_AQ               TO strmadmin;
GRANT EXECUTE ON DBMS_AQADM            TO strmadmin;
GRANT EXECUTE ON DBMS_CAPTURE_ADM      TO strmadmin;
GRANT EXECUTE ON DBMS_FLASHBACK        TO strmadmin;
GRANT EXECUTE ON DBMS_STREAMS_ADM      TO strmadmin;BEGIN DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(privilege    => DBMS_RULE_ADM.CREATE_RULE_SET_OBJ, grantee      => 'strmadmin', grant_option => FALSE);
END;
/BEGIN DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(privilege    => DBMS_RULE_ADM.CREATE_RULE_OBJ, grantee      => 'strmadmin', grant_option => FALSE);
END;
/

--建立流队列，名称叫streams_queue ，用于存储捕获的变化

CONNECT strmadmin/strmadmin
EXEC DBMS_STREAMS_ADM.SET_UP_QUEUE();

--配置logmnr使用的表空间，我们就用tools

conn / as sysdba
EXECUTE DBMS_LOGMNR_D.SET_TABLESPACE('TOOLS');

--增强日志的模式

ALTER TABLE hr.employees  ADD SUPPLEMENTAL LOG GROUP log_group_employees_pk(employee_id) ALWAYS;

--配置捕获程序

CONNECT strmadmin/strmadmin
BEGINDBMS_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);
END;
/

--设置scn

DECLAREiscn  NUMBER;         -- Variable to hold instantiation SCN value
BEGINiscn := DBMS_FLASHBACK.GET_SYSTEM_CHANGE_NUMBER();DBMS_APPLY_ADM.SET_TABLE_INSTANTIATION_SCN(source_object_name    => 'hr.employees',source_database_name  => 'ora9',instantiation_scn     => iscn);
END;
/

--配置叫emp_agent的代理程序

BEGINDBMS_AQADM.DROP_AQ_AGENT(agent_name => 'emp_agent');
END;
/BEGINDBMS_AQADM.CREATE_AQ_AGENT(agent_name => 'emp_agent');DBMS_AQADM.ENABLE_DB_ACCESS(agent_name  => 'emp_agent',db_username => 'strmadmin');
END;
/

--建立队列订户

DECLAREsubscriber SYS.AQ$_AGENT;
BEGINsubscriber :=  SYS.AQ$_AGENT('emp_agent', NULL, NULL);  SYS.DBMS_AQADM.ADD_SUBSCRIBER(queue_name          =>  'strmadmin.streams_queue',subscriber                  =>  subscriber,rule                =>  NULL,transformation      =>  NULL);
END;
/

--建立存储过程enq_row_lcr

CREATE OR REPLACE PROCEDURE enq_row_lcr(in_any IN SYS.ANYDATA) ISenqopt       DBMS_AQ.ENQUEUE_OPTIONS_T;mprop        DBMS_AQ.MESSAGE_PROPERTIES_T;recipients   DBMS_AQ.AQ$_RECIPIENT_LIST_T;enq_eventid  RAW(16);
BEGINmprop.SENDER_ID := SYS.AQ$_AGENT(name     => 'emp_agent',address  => NULL,protocol => NULL);recipients(1) := SYS.AQ$_AGENT(name     => 'emp_agent',address  => NULL,protocol => NULL);mprop.RECIPIENT_LIST := recipients;DBMS_AQ.ENQUEUE(queue_name         => 'strmadmin.streams_queue',enqueue_options    => enqopt,message_properties => mprop,payload            => in_any,msgid              => enq_eventid);
END;
/

--建立DML处理存储过程

CREATE OR REPLACE PROCEDURE emp_dml_handler(in_any IN SYS.ANYDATA) ISlcr          SYS.LCR$_ROW_RECORD;rc           PLS_INTEGER;command      VARCHAR2(10);old_values   SYS.LCR$_ROW_LIST;
BEGIN-- Re-enqueue the row LCR for explicit dequeue by another applicationenq_row_lcr(in_any);-- Access the LCRrc := in_any.GETOBJECT(lcr);-- Get the object command typecommand := lcr.GET_COMMAND_TYPE();-- Check for DELETE command on the hr.employees tableIF command = 'DELETE' THEN-- Set the command_type in the row LCR to INSERTlcr.SET_COMMAND_TYPE('INSERT');-- Set the object_name in the row LCR to EMP_DELlcr.SET_OBJECT_NAME('EMP_DEL');-- Get the old values in the row LCRold_values := lcr.GET_VALUES('old');-- Set the old values in the row LCR to the new values in the row LCRlcr.SET_VALUES('new', old_values);-- Set the old values in the row LCR to NULLlcr.SET_VALUES('old', NULL);-- Add a SYSDATE value for the timestamp columnlcr.ADD_COLUMN('new', 'TIMESTAMP', SYS.AnyData.ConvertDate(SYSDATE));--  Apply the row LCR as an INSERT into the EMP_DEL tablelcr.EXECUTE(true);END IF;
END;
/

--配置DML管理者，为hr.employees

BEGINDBMS_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);
END;
/BEGINDBMS_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);
END;
/BEGINDBMS_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);
END;
/

--建立存储过程为出列和再入列事件

CREATE OR REPLACE PROCEDURE emp_dq (consumer IN VARCHAR2) ASdeqopt       DBMS_AQ.DEQUEUE_OPTIONS_T;mprop        DBMS_AQ.MESSAGE_PROPERTIES_T;msgid        RAW(16);payload      SYS.AnyData;new_messages BOOLEAN := TRUE;row_lcr      SYS.LCR$_ROW_RECORD;tc           pls_integer;next_trans   EXCEPTION;no_messages  EXCEPTION; pragma exception_init (next_trans, -25235);pragma exception_init (no_messages, -25228);
BEGINdeqopt.consumer_name := consumer;deqopt.wait := 1;WHILE (new_messages) LOOPBEGINDBMS_AQ.DEQUEUE(queue_name          =>  'strmadmin.streams_queue',dequeue_options     =>  deqopt,message_properties  =>  mprop,payload             =>  payload,msgid               =>  msgid);COMMIT;deqopt.navigation := DBMS_AQ.NEXT;IF (payload.GetTypeName = 'SYS.LCR$_ROW_RECORD') THENtc := payload.GetObject(row_lcr);   DBMS_OUTPUT.PUT_LINE(row_lcr.GET_COMMAND_TYPE || ' row LCR dequeued');END IF;                       EXCEPTIONWHEN next_trans THENdeqopt.navigation := DBMS_AQ.NEXT_TRANSACTION;WHEN no_messages THENnew_messages  := FALSE;DBMS_OUTPUT.PUT_LINE('No more events');END;END LOOP; 
END;
/

--配置应用程序

BEGINDBMS_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 => 'ora9');
END;
/

--启动应用程序

BEGINDBMS_APPLY_ADM.SET_PARAMETER(apply_name  => 'apply_emp', parameter   => 'disable_on_error', value       => 'n');
END;
/BEGINDBMS_APPLY_ADM.START_APPLY(apply_name  => 'apply_emp');
END;
/

--启动捕获程序

BEGINDBMS_CAPTURE_ADM.START_CAPTURE(capture_name  => 'capture_emp');
END;
/--对hr.employees进行插入，删除和修改
conn hr/hr
INSERT INTO hr.employees values(207, 'JOHN', 'SMITH', 'JSMITH@MYCOMPANY.COM', NULL, '07-JUN-94', 'AC_ACCOUNT', 777, NULL, NULL, 110);
COMMIT;UPDATE hr.employees SET salary=5999 WHERE employee_id=206;
COMMIT;DELETE FROM hr.employees WHERE employee_id=207;
COMMIT;CONNECT strmadmin/strmadmin
SELECT * FROM hr.emp_del;SELECT MSG_ID, MSG_STATE, CONSUMER_NAME FROM AQ$STREAMS_QUEUE_TABLE;EXEC emp_dq('emp_agent');SELECT MSG_ID, MSG_STATE, CONSUMER_NAME FROM AQ$STREAMS_QUEUE_TABLE;

--显示应用程序的错误

COLUMN APPLY_NAME HEADING 'Apply|Process|Name' FORMAT A8
COLUMN SOURCE_DATABASE HEADING 'Source|Database' FORMAT A8
COLUMN LOCAL_TRANSACTION_ID HEADING 'Local|Transaction|ID' FORMAT A11
COLUMN ERROR_MESSAGE HEADING 'Error Message' FORMAT A50SELECT APPLY_NAME, SOURCE_DATABASE, LOCAL_TRANSACTION_ID, ERROR_MESSAGEFROM DBA_APPLY_ERROR;