环境：
源数据库：oracle 9.2.0.8
目标数据库：oracle 10.2.0.3
跨版本的stream配置，结果会向低版本兼容。
STREAMS 安装
-------------
安装分以下4步：

1、设置与streams相关的参数
2、设置目标数据
3、设置源数据库
4、用exp/imp从源库导出的原始数据表，导入到目标库。

第一部分
跟streams相关的参数
---------------------------------------------
1.1 COMPATIBLE :
在streams中，conpatible必须设置大于等于9.2.0

1.2 GLOBAL_NAMES :
该参数必须设置为true
该参数设置后就不要修改，stream的capture、propagation、apply都会用到源数据库的全局名，capture在自动捕获的时候会自动用到源库的全局名。
如果global_names一定要修改，要在没有用户修改数据的时候。这时streams需要重新配置。

1.3 JOB_QUEUE_PROCESSES :
这个参数用于设置可以处理dbms_job请求的进程数，确定其设置大于等于2

1.4 AQ_TM_PROCESSES :
设置该参数大于等于1，用于启动监控进程队列数。

1.5 LOGMNR_MAX_PERSISTENT_SESSIONS :
这个参数用于设置最大固定的logminer的挖掘会话数。
streams的捕获进程通过logminer去挖掘重做日志。
如果在某个数据库需要运行多个stream捕获进程，那么这个参数需要设置为大于或等于计划需要运行的捕获进程数。

1.6 LOG_PARALLELISM :
该参数必须在需要捕获事件的数据库上设置为1

1.7 PARALLEL_MAX_SERVERS :
每个捕获进程和应用进程可能用到多个并行执行服务器。
应用进程默认需要两个并行服务器，所以对单个串行的应用进程这个参数至少需要设置为2。
设置该参数合适的值以确保有足够的并行执行服务器。


1.8 SHARED_POOL_SIZE :
每个捕获进程需要10MB shared pool的空间，但stream被限制为最多只能利用shared pool 10%的空间。
所以shared_pool_size最小必须设置为100MB，如果需要使用更多的捕获进程，则shared_pool_size需要设置为一个足够高的值。

1.9 OPEN_LINKS :
在一个远程会话中最大的并行连接数，确保该参数大于等于4

1.10 streams中的数据库处于归档模式。（目标库可以是非归档））

第二部分
目标库配置（info）
----------------------------------------------------------------------------
2.1 创建streams管理用户stradmin:
connectSYS/SYS01@STREAMDas SYSDBA
create user STRADMIN identified by STRADMIN default tablespace stream_ts;

2.2 赋予streams管理用户stradmin足够的管理权限
GRANT CONNECT, RESOURCE, AQ_ADMINISTRATOR_ROLE,DBA to STRADMIN;
GRANT SELECT ANY DICTIONARY TO STRADMIN;
GRANT EXECUTE ON DBMS_AQ TO STRADMIN;
GRANT EXECUTE ON DBMS_AQADM TO STRADMIN;
GRANT EXECUTE ON DBMS_FLASHBACK TO STRADMIN;
GRANT EXECUTE ON DBMS_STREAMS_ADM TO STRADMIN;
GRANT EXECUTE ON DBMS_CAPTURE_ADM TO STRADMIN;
GRANT EXECUTE ON DBMS_APPLY_ADM TO STRADMIN;
GRANT EXECUTE ON DBMS_RULE_ADM TO STRADMIN;
GRANT EXECUTE ON DBMS_PROPAGATION_ADM TO STRADMIN;

--以下权限可选
BEGIN
DBMS_AQADM.GRANT_SYSTEM_PRIVILEGE(
privilege => 'ENQUEUE_ANY',
grantee => 'STRADMIN',
admin_option => FALSE);
END;
/

BEGIN
DBMS_AQADM.GRANT_SYSTEM_PRIVILEGE(
privilege => 'DEQUEUE_ANY',
grantee => 'STRADMIN',
admin_option => FALSE);
END;
/

BEGIN
DBMS_AQADM.GRANT_SYSTEM_PRIVILEGE(
privilege => 'MANAGE_ANY',
grantee => 'STRADMIN',
admin_option => TRUE);
END;
/

BEGIN
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.CREATE_EVALUATION_CONTEXT_OBJ,
grantee => 'STRADMIN',
grant_option => TRUE);
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.CREATE_RULE_SET_OBJ,
grantee => 'STRADMIN',
grant_option => TRUE);
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.CREATE_RULE_OBJ,
grantee => 'STRADMIN',
grant_option => TRUE);
END;
/

BEGIN
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.CREATE_ANY_RULE_SET,
grantee => 'STRADMIN',
grant_option => TRUE);
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.ALTER_ANY_RULE_SET,
grantee => 'STRADMIN',
grant_option => TRUE);
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.EXECUTE_ANY_RULE_SET,
grantee => 'STRADMIN',
grant_option => TRUE);
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.CREATE_ANY_RULE,
grantee => 'STRADMIN',
grant_option => TRUE);
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.ALTER_ANY_RULE,
grantee => 'STRADMIN',
grant_option => TRUE);
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.EXECUTE_ANY_RULE,
grantee => 'STRADMIN',
grant_option => TRUE);
END;
/

BEGIN
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege => DBMS_RULE_ADM.EXECUTE_ANY_EVALUATION_CONTEXT,
grantee => 'STRADMIN',
grant_option => TRUE);
END;
/

2.3在streams管理用户下 创建streams队列:

connectSTRADMIN/STRADMIN@STREAMD
----创建连接到源库的db link
CREATE DATABASE LINK stream connect to STRADMIN identified by STRADMIN using 'stream';

BEGIN
DBMS_STREAMS_ADM.SET_UP_QUEUE(
queue_table => 'STREAMS_QUEUE_TABLE',
queue_name => 'STREAMS_QUEUE',
queue_user => 'STRADMIN');
END;
/

BEGIN
DBMS_STREAMS_ADM.REMOVE_QUEUE(
queue_name => 'STREAMS_QUEUE',
cascade =>TRUE
);
END;
/

2.4 在目标库上添加/删除应用规则，可以基于表/用户
----ADD TABLE RULE（基于表的应用）
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name => 'TEST.EMP',
streams_type => 'APPLY',
streams_name => 'STRADMIN_APPLY',
queue_name => 'STRADMIN.STREAMS_QUEUE',
include_dml => true,
include_ddl => true,
source_database => 'STREAM');
END;
/

---REMOVE TABLE RULE
BEGIN
DBMS_STREAMS_ADM.REMOVE_RULE(
rule_name =>'&rule_name',
streams_type => 'APPLY',
streams_name => 'STRADMIN_APPLY');
END;
/

BEGIN
DBMS_STREAMS_ADM.REMOVE_RULE(
rule_name =>'DEPT27',
streams_type => 'APPLY',
streams_name => 'STRADMIN_APPLY');
END;
/

DBMS_STREAMS_ADM.REMOVE_STREAMS_CONFIGURATION; ---在10g中可以用

---ADD SCHEMA RULE（基于schema的应用）
BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
schema_name =>'TEST',
streams_type =>'APPLY',
streams_name =>'STRADMIN_APPLY',
queue_name => 'STRADMIN.STREAMS_QUEUE',
include_dml => TRUE,
include_ddl =>TRUE,
source_database =>'STREAM'
);
END;
/

2.5 在目标库上说明APPLY USER的用户
这个用户将在目标库上apply所有的dml和ddl操作，所以该用户（apply_user）必须具有在应用对象上执行dml和ddl的权限。
BEGIN
DBMS_APPLY_ADM.ALTER_APPLY(
apply_name => 'STRADMIN_APPLY',
apply_user => 'STRADMIN');
END;
/

2.6 如果不希望应用进程因为出现错误而异常中断，可以设置如下参数。
默认值"Y"表示任何错误将终止应用进程。
当设置为“N”，应用进程不会因为错误而终止，错误信息会记录到dba_apply_error表中。

BEGIN
DBMS_APPLY_ADM.SET_PARAMETER(
apply_name => 'STRADMIN_APPLY',
parameter => 'DISABLE_ON_ERROR',
value => 'N' );
END;
/

2.7 启动/停止应用进程
BEGIN
DBMS_APPLY_ADM.START_APPLY(apply_name => 'STRADMIN_APPLY');
END;
/

BEGIN
DBMS_APPLY_ADM.STOP_APPLY(apply_name => 'STRADMIN_APPLY');
END;
/

第三部分
源库配置（stream）
--------------------------------------------------------------------
3.1 把logminer表从system表空间中移走
默认情况下，所有的logminer表都在system表空间中，为logminer表创建一个单独的表空间是一个好的习惯。

CREATE TABLESPACE LOGMNRTS DATAFILE 'F:ORADATAlogmnrts.dbf' SIZE 100M ;

BEGIN
DBMS_LOGMNR_D.SET_TABLESPACE('LOGMNRTS');
END;
/

3.2 打开dept表的 supplemental logging 功能。

connectSYSSYS01@STREAMSas SYSDBA
ALTER TABLE TEST.EMP ADD SUPPLEMENTAL LOG GROUP log1_pk (DEPTNO) ALWAYS;

ALTER DATABASE ADD SUPPlEMENTAL LOG DATA(primary key,unique index) columns;

3.3 创建streams管理用户和赋予相应的权限（stradmin)
重复2.1和2.2的步骤

3.4创建到目标库的db link :
connectSTRADMIN/STRADMIN@STREAMS
CREATE DATABASE LINK info connect to STRADMIN identified by STRADMIN using 'info';

测试db link，访问目标库是否成功
Eg : select * fromglobal_name@info;

3.5 创建stream队列
BEGIN
DBMS_STREAMS_ADM.SET_UP_QUEUE(
queue_name => 'STREAMS_QUEUE',
queue_table =>'STREAMS_QUEUE_TABLE',
queue_user => 'STRADMIN');
END;
/
---删除、停止、启动queue
exec DBMS_STREAMS_ADM.REMOVE_QUEUE( 'STREAMS_QUEUE');
exec DBMS_AQADM.STOP_QUEUE('STREAMS_QUEUE');
exec DBMS_AQADM.DROP_QUEUE ('STREAMS_QUEUE');
exec DBMS_AQADM.START_QUEUE('STREAMS_QUEUE');

3.6 在源库上添加捕获表的规则

BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name => 'TEST.EMP',
streams_type => 'CAPTURE',
streams_name => 'STRADMIN_CAPTURE',
queue_name => 'STRADMIN.STREAMS_QUEUE',
include_dml => true,
include_ddl => true,
source_database => 'STREAM');
END;
/

---在源库上添加捕获schema的规则
BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
schema_name =>'TEST',
streams_type =>'CAPTURE',
streams_name =>'STRADMIN_CAPTURE',
queue_name => 'STRADMIN.STREAMS_QUEUE',
include_dml => TRUE,
include_ddl =>TRUE,
source_database =>'STREAM'
);
END;
/

---删除规则
BEGIN
DBMS_STREAMS_ADM.REMOVE_RULE(
rule_name =>'EMP51',
streams_type =>'CAPTURE',
streams_name =>'STRADMIN_CAPTURE'
);

DBMS_STREAMS_ADM.REMOVE_RULE(
rule_name =>'EMP52',
streams_type =>'CAPTURE',
streams_name =>'STRADMIN_CAPTURE'
);

DBMS_STREAMS_ADM.REMOVE_RULE(
rule_name =>'EMP49',
streams_type =>'PROPAGATION',
streams_name =>'STRADMIN_PROPAGATE'
);

DBMS_STREAMS_ADM.REMOVE_RULE(
rule_name =>'EMP50',
streams_type =>'PROPAGATION',
streams_name =>'STRADMIN_PROPAGATE'
);
END;
/

3.7在源库上添加传播表的规则
该步也会在目标库上创建一个传播任务

BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(
table_name => 'TEST.EMP',
streams_name => 'STRADMIN_PROPAGATE',
source_queue_name => 'STRADMIN.STREAMS_QUEUE',
destination_queue_name =>'STRADMIN.STREAMS_QUEUE@INFO',
include_dml => true,
include_ddl => true,
source_database => 'STREAM');
END;
/

----在源库上创建一个传播schema的规则。
BEGIN
DBMS_STREAMS_ADM.ADD_SCHEMA_PROPAGATION_RULES(
schema_name =>'TEST',
streams_name =>'STRADMIN_PROPAGATE',
source_queue_name =>'STRADMIN.STREAMS_QUEUE',
destination_queue_name =>'STRADMIN.STREAMS_QUEUE@INFO',
include_dml => true,
include_ddl => true,
source_database => 'STREAM');
END;
/

BEGIN
DBMS_PROPAGATION_ADM.START_PROPAGATION('PROPAGATION_NAME');
END;
/

第四部分
从源库到目标库导出，导入，实例化表/schema
------------------------------------------------------------------------------
4.1 如果对象在目标库中不存在，在源库导出对象并在目标库导入对象。

源库导出：
在导出命令中设置OBJECT_CONSISTENT=Y ，保证每个独立的对象在某个系统改变号SCN在导出时是一致的。

在9上export
exp USERID=SYSTEM/SYS01 TABLES=TEST.EMP FILE=D: ests.dmp OBJECT_CONSISTENT=Y INDEXES=Y STATISTICS = NONE GRANTS=Y ROWS=Y
exp USERID=SYSTEM/SYS01 ōwner=test FILE=D: ests.dmp OBJECT_CONSISTENT=Y INDEXES=Y STATISTICS = NONE GRANTS=Y ROWS=Y

目标库导入：
在导入命令中设置STREAMS_INSTANTIATION=Y，以便保证目标库上的stream元数据根据导出文件中的SCN被更新。

imp USERID=system/sys01 FILE=D: ests.dmp fromuser=test touser=test CONSTRAINTS=Y IGNORE=Y GRANTS=Y ROWS=Y COMMIT=Y STREAMS_INSTANTIATION=Y


4.2 如果所有对象在目标库中存在，有两种方法可以在目标库上实例化对象。

1. 只导入/导出元数据 :

设置ROWS=N导出源数据库
expUSERID=SYSTEM@STREAMsōWNER=TEST FILE=D: est_tables.dmp ROWS=N OBJECT_CONSISTENT=Y

设置IGNORE=Y导入目标数据库
impUSERID=SYSTEM@streamdFULL=Y FILE=/share/test_tables.dmp IGNORE=Y STREAMS_INSTANTIATION=Y

2. 手工实例化所有对象
在源库上取得实例化scn号:
connectSTRADMIN/STRADMIN@STREAMs
set serveroutput on
DECLARE
iscn NUMBER; -- Variable to hold instantiation SCN value
BEGIN
iscn := DBMS_FLASHBACK.GET_SYSTEM_CHANGE_NUMBER();
DBMS_OUTPUT.PUT_LINE ('Instantiation SCN is: ' || iscn);
END;

用该SCN实例化目标库上的对象
过程 SET_TABLE_INSTANTIATION_SCN控制表的那些LCRs将被应用进程应用。
如果源库上LCRs的提交scn小于等于实例化的scn，则应用进程丢弃该LCR，否则应用进程应用该scn。

connectSTRADMIN/STRADMIN@streamd
BEGIN
DBMS_APPLY_ADM.SET_TABLE_INSTANTIATION_SCN(
source_object_name => 'TEST.EMP',
source_database_name =>' STREAM',
instantiation_scn => 18902052364 );
END;

DBMS_APPLY_ADM.SET_SCHEMA_INSTANTIATION_SCN('TEST','STREAM', 18902294194 );

-----------------------------------------------------------------------------
最后在源库上启动捕获进程：

connectSTRADMIN/STRADMIN@STREAMs
BEGIN
DBMS_CAPTURE_ADM.START_CAPTURE(capture_name => 'STRADMIN_CAPTURE');
END;
/
---停止捕获
BEGIN
DBMS_CAPTURE_ADM.STOP_CAPTURE(capture_name => 'STRADMIN_CAPTURE');
END;
/

现在可以在两个数据库中通过stream复制数据了