Oracle  流stream数据的复制

--实验的目的是捕获scott.emp1表的变化，将变化应用到远程数据库scott.emp1表中。
--设置初始化参数
AQ_TM_PROCESSES=1
COMPATIBLE=9.2.0
LOG_PARALLELISM=1
GLOBAL_NAMES=true
JOB_QUEUE_PROCESSES=2
--查看数据库的名称，我的为ora9,将以下的ora9全部替换为你的数据库名称
--数据库为归档模式
conn scott/tiger
drop table emp1;
create table emp1 as select * from emp;
alter table emp1 add constraint pk_emp1 primary key (empno);

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

create tablespace streamout datafile 'F:\ORACLE\ORADATA\ORA9\streamout.dbf' size 20m
autoextend on;

create tablespace streamin datafile 'F:\ORACLE\ORADATA\ORA9\streamin.dbf' size 20m
autoextend on;

create tablespace tslogmnr datafile 'F:\ORACLE\ORADATA\ORA9\logmnr.dbf' size 20m
autoextend on;

drop user streamout cascade;
drop user streamin cascade;

--建立streamout帐号-------------------------------------------------------------------
conn / as sysdba

GRANT CONNECT, RESOURCE, SELECT_CATALOG_ROLE 
  TO streamout IDENTIFIED BY streamout;
ALTER USER streamout DEFAULT TABLESPACE streamout;

GRANT EXECUTE ON DBMS_AQADM            TO streamout;
GRANT EXECUTE ON DBMS_CAPTURE_ADM      TO streamout;
GRANT EXECUTE ON DBMS_PROPAGATION_ADM  TO streamout;
GRANT EXECUTE ON DBMS_STREAMS_ADM      TO streamout;

BEGIN 
  DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
    privilege    => DBMS_RULE_ADM.CREATE_RULE_SET_OBJ, 
    grantee      => 'streamout', 
    grant_option => FALSE);
END;
/

BEGIN 
  DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
    privilege    => DBMS_RULE_ADM.CREATE_RULE_OBJ, 
    grantee      => 'streamout', 
    grant_option => FALSE);
END;
/

--建立流队列，名称叫streams_queue ，用于存储捕获的变化,同时建立了表streams_queue_table
CONNECT streamout/streamout
EXEC DBMS_STREAMS_ADM.SET_UP_QUEUE();
select * from tab;

--建立数据库连接,连接到远程，先配置网络连接，建立的时候要使用数据库的名称
CREATE DATABASE LINK ora11 CONNECT TO streamin IDENTIFIED BY streamin   USING 'ora11';

--建立streamin帐号----要是远程其它数据库-------------------------
conn / as sysdba
GRANT CONNECT, RESOURCE, SELECT_CATALOG_ROLE 
  TO streamin IDENTIFIED BY streamin;
ALTER USER streamin DEFAULT TABLESPACE streamin;

GRANT EXECUTE ON DBMS_APPLY_ADM        TO streamin;
GRANT EXECUTE ON DBMS_AQADM            TO streamin;
GRANT EXECUTE ON DBMS_STREAMS_ADM      TO streamin;

BEGIN 
  DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
    privilege    => DBMS_RULE_ADM.CREATE_RULE_SET_OBJ, 
    grantee      => 'streamin', 
    grant_option => FALSE);
END;
/

BEGIN 
  DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
    privilege    => DBMS_RULE_ADM.CREATE_RULE_OBJ, 
    grantee      => 'streamin', 
    grant_option => FALSE);
END;
/

conn streamin/streamin
EXEC  DBMS_STREAMS_ADM.SET_UP_QUEUE();
select * from tab;

----------------------------本地的数据库--------------------------
--配置logmnr使用的表空间，我们就用tslogmnr
conn / as sysdba
EXECUTE DBMS_LOGMNR_D.SET_TABLESPACE('tslogmnr');

--增强日志的模式
ALTER TABLE scott.emp1  ADD SUPPLEMENTAL LOG GROUP log_group_emp1_pk (empno) ALWAYS;
grant all on scott.emp1 to streamout;

--配置捕获城序-----------------输出的帐号-----------------
CONNECT streamout/streamout
BEGIN
  DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(
    table_name              => 'scott.emp1', 
    streams_name            => 'str1_to_str2', 
    source_queue_name       => 'streamout.streams_queue',
    destination_queue_name  => 'streamin.streams_queue@ora11',
    include_dml             =>  true,
    include_ddl             =>  true,
    source_database         => 'ora9');
END;
/

BEGIN
  DBMS_STREAMS_ADM.ADD_TABLE_RULES(
    table_name     => 'scott.emp1',   
    streams_type   => 'capture',
    streams_name   => 'capture_simp',
    queue_name     => 'streamout.streams_queue',
    include_dml    =>  true,
    include_ddl    =>  true);
END;
/

-------------------------------dos下exp/imp----------------------------------
--实例化emp1表，emp1存在于scott帐号内
将scott.emp1导出，再导入到远程数据库的scott帐号
exp scott/tiger FILE=d:\bk\1.dmp TABLES=emp1 OBJECT_CONSISTENT=y ROWS=n
--imp到远程数据库中
imp scott/tiger FILE=d:\bk\1.dmp IGNORE=y COMMIT=y LOG=import.log STREAMS_INSTANTIATION=y 
-----------------------------远程数据库---------------------------------
conn scott/tiger
ALTER TABLE emp1 DROP SUPPLEMENTAL LOG GROUP log_group_emp1_pk;
GRANT ALL ON EMP1 TO streamin;

--配置应用程序-------------------输入的帐号------远程数据库----------
conn streamin/streamin
BEGIN
  DBMS_STREAMS_ADM.ADD_TABLE_RULES(
    table_name      => 'scott.emp1',
    streams_type    => 'apply', 
    streams_name    => 'apply_simp',
    queue_name      => 'streamin.streams_queue',
    include_dml     =>  true,
    include_ddl     =>  true,
    source_database => 'ora9');
END;
/

--启动应用程序
conn streamin/streamin
BEGIN
  DBMS_APPLY_ADM.SET_PARAMETER(
    apply_name  => 'apply_simp', 
    parameter   => 'disable_on_error', 
    value       => 'n');
END;
/

conn streamin/streamin
BEGIN
  DBMS_APPLY_ADM.START_APPLY(
    apply_name  => 'apply_simp');
END;
/

--如果想停止应用，该句话不运行
conn streamin/streamin
BEGIN
  DBMS_APPLY_ADM.STOP_APPLY(
    apply_name  => 'apply_simp');
END;
/
--启动捕获程序----------------------输出的帐号-----本地数据库-------
conn streamout/streamout
--如果想停止应用，该句话不运行
BEGIN
  DBMS_CAPTURE_ADM.STop_CAPTURE(
    capture_name  => 'capture_simp');
END;
/
--启动
conn streamout/streamout
BEGIN
  DBMS_CAPTURE_ADM.START_CAPTURE(
    capture_name  => 'capture_simp');
END;
/

--对scott.emp1进行插入，删除和修改
conn scott/tiger
update emp1 set sal=300 where empno=7369;
COMMIT;

--远程数据库
CONNECT scott/tiger
SELECT * FROM emp1;

--显示应用程序的错误------------------------------------------------------
conn streamin/streamin
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 A50

SELECT APPLY_NAME, SOURCE_DATABASE, LOCAL_TRANSACTION_ID, ERROR_MESSAGE
  FROM DBA_APPLY_ERROR;

conn / as sysdba

select * from DBA_QUEUE_SCHEDULES;

SELECT 
  p.SOURCE_QUEUE_OWNER||'.'||
    p.SOURCE_QUEUE_NAME||'@'||
    g.GLOBAL_NAME SOURCE_QUEUE, 
  p.DESTINATION_QUEUE_OWNER||'.'||
    p.DESTINATION_QUEUE_NAME||'@'||
    p.DESTINATION_DBLINK DESTINATION_QUEUE
  FROM DBA_PROPAGATION p, GLOBAL_NAME g;
SELECT STATUS FROM DBA_CAPTURE;

select * from V$STREAMS_CAPTURE ;