一、STREAM环境准备：
This chapter contains these topics:
 Configuring a Streams Administrator
 Setting Initialization Parameters Relevant to Streams
 Configuring Network Connectivity and Database Links
1.创建stream管理用户; ---或都创建一个单独的用户,或使用已存在的用户并赋予相应的权限 不能使用sys、system且不能使用system tablespace作为默认表空间
主：
CREATE TABLESPACE streams_tbs DATAFILE 'C:\ORACLE\PRODUCT\10.2.0\ORADATA\ORA10G\streams_tbs.dbf'  SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
CREATE USER strmadmin IDENTIFIED BY strmadminpw DEFAULT TABLESPACE streams_tbs QUOTA UNLIMITED ON streams_tbs;
grant connect,resource,dba,aq_administrator_role to strmadmin; 

BEGIN
DBMS_STREAMS_AUTH.GRANT_ADMIN_PRIVILEGE(
grantee          => 'strmadmin',    
grant_privileges => true);
END;
/
备：
CREATE TABLESPACE streams_tbs DATAFILE '/oradata/shrnc/streams_tbs.dbf'  SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
CREATE USER strmadmin IDENTIFIED BY strmadminpw DEFAULT TABLESPACE streams_tbs QUOTA UNLIMITED ON streams_tbs;
grant connect,resource,dba,aq_administrator_role to strmadmin; 
BEGIN
DBMS_STREAMS_AUTH.GRANT_ADMIN_PRIVILEGE(
grantee          => 'strmadmin',    
grant_privileges => true);
END;
/
2.设置初始化参数对stream的操作、安全、性能方面都非常重要，通过alter system或alter session设置合适的参数（结合官方文档modifialbe来判断是否修改）
Sqlplus ‘/ as sysdba’ 
alter system set aq_tm_processes=2 scope=both; 
alter system set global_names=true scope=both; 
alter system set job_queue_processes=10 scope=both; 
alter system set parallel_max_servers=20 scope=both; 
alter system set undo_retention=3600 scope=both; 
alter system set nls_date_format='YYYY-MM-DD HH24:MI:SS' scope=spfile; 
alter system set streams_pool_size=25M scope=spfile; 
alter system set utl_file_dir='*' scope=spfile; 
alter system set open_links=4 scope=spfile;
执行完毕后重启数据库。
3.配合网络连接（TNS)和Dblink
3.1主数据库(tnsnames.ora)中添加从数据库的配置。
以下是引用片段：
#Oracle stream master node  
master =
  (DESCRIPTION =
    (ADDRESS_LIST =
      (ADDRESS = (PROTOCOL = TCP)(HOST = 192.168.1.5)(PORT = 1521))
    )
    (CONNECT_DATA =
      (SERVICE_NAME = ora10g)
    )
  )
3.2配置从环境tnsnames.ora
以下是引用片段：
　　从数据库(tnsnames.ora)中添加主数据库的配置。 
#Oracle stream salve node
slave =
  (DESCRIPTION =
    (ADDRESS_LIST =
      (ADDRESS = (PROTOCOL = TCP)(HOST = 192.168.1.104)(PORT = 1521))
    )
    (CONNECT_DATA =
      (SERVICE_NAME = shsnc)
    )
  )
3.3 Dblink 创建
主：
conn strmadmin/strmadminpw
create database link shrnc connect to strmadmin identified by "strmadminpw" using 'slave';
备：
conn strmadmin/strmadminpw
create database link ora10g connect to strmadmin identified by "strmadminpw" using 'master';
配置数据库去运行捕获进程：
The following sections describe database requirements for running a Streams
capture process:
Configuring the Database to Run in ARCHIVELOG Mode
Specifying an Alternate Tablespace for LogMiner
In addition to these tasks, make sure the initialization parameters are set properly
on any database that will run a capture process.
1.将数据库置为归档模式
设置log_archive_dest_1到相应的位置;设定log_archive_start为TRUE，即启用自动归档功能;设定log_archive_format指定归档日志的命令格式。
举例：
以下是引用片段：
sqlplus '/ as sysdba' 
alter system set log_archive_dest_1='location=C:\ORACLE\PRODUCT\10.2.0\ORADATA\ORA10G' scope=spfile; 
alter system set log_archive_start=TRUE scope=spfile; 
alter system set log_archive_format='arch%t_%s_%r.arc' scope=spfile; 
shutdown immediate; 
startup mount; 
alter database archivelog; 
alter database open;
数据库置为归档模式后，可以按如下方式检验一下：
以下是引用片段：
SQL> archive log list 
2.指定一个表空间为logminer
默认logminer表在system表空间上，但system表空间可能没有足够的空间；
execute dbms_logmnr_d.set_tablespace('streams_tbs'); 
二、创建Capture进程
（如果增加要配置的表，直接在STRMADMIN.TMP_STREAM_TABLES中添加即可，并且批量脚本的执行需要在中心库执行）
以下是引用片段：
#以strmadmin身份，登录主数据库。提醒一下，本文档以表hurp.test表做示例。 
connect strmadmin/strmadminpw 
SQL> create table TMP_STREAM_TABLES(owner varchar2(20),table_name varchar2(30));
表已创建。             
SQL> insert into TMP_STREAM_TABLES values('HURP','TEST');                       
已创建 1 行。          
SQL> commit;                                                                    
提交完成。
（如果增加要配置的表，直接在STRMADMIN.TMP_STREAM_TABLES中添加即可，并且批量脚本的执行需要在中心库执行）
SELECT 'begin dbms_streams_adm.add_table_rules(table_name => '''||OWNER||'.'||
       TABLE_NAME || ''',' || 'streams_type => ''capture'',' ||
       'streams_name => ''CAPTURE_CENDB1'',' ||
       'queue_name => ''slave_queue'',' ||
       'include_dml => true,' || 'include_ddl => true,' ||
       'inclusion_rule => true); end; /'
  FROM TMP_STREAM_TABLES;

begin 
dbms_streams_adm.add_table_rules(
table_name => 'HURP.TEST',
streams_type => 'capture',
streams_name => 'capture_master',
queue_name => 'strmadmin.master_queue',
include_dml => true,
include_ddl => true,
inclusion_rule => true); 
end;
/
begin 
dbms_streams_adm.remove_table(
streams_type => 'capture',
streams_name => 'CAPTURE_MASTER');
end;
/
三、实例化复制数据库
在主数据库环境中，执行如下Shell语句。如果从库的admin用户不存在，建立一个admin的空用户。
exp  file=C:\ORACLE\PRODUCT\10.2.0\ORADATA\ORA10G\test.dmp object_consistent=y rows=y tables=hurp.test
imp file=/oradata/arch/test.dmp  ignore=y commit=y  streams_instantiation=y  fromuser=hurp touser=hurp

三、管理Staging和Propagation
This chapter contains these topics:
n Managing Streams Queues
n Managing Streams Propagations and Propagation Jobs
n Managing a Streams Messaging Environment
创建流队列：
1.创建Master流队列
以下是引用片段：、
#以strmadmin身份，登录主数据库。 
connect strmadmin/strmadminpw 
begin 
dbms_streams_adm.set_up_queue( 
queue_table => 'master_queue_table', 
queue_name => 'master_queue'); 
end; 
/
begin
dbms_streams_adm.REMOVE_QUEUE( 
queue_name => 'slave_queue'); 
end; 
/
2.创建slave流队列
以下是引用片段：
#以strmadmin身份，登录从数据库。 
connect strmadmin/strmadminpw 
begin 
dbms_streams_adm.set_up_queue( 
queue_table => 'slave_queue_table', 
queue_name => 'slave_queue'); 
end; 
/


begin
dbms_streams_adm.REMOVE_QUEUE( 
queue_name => 'slave_queue'); 
end; 
/
创建传播进程：
以下是引用片段：
connect strmadmin/strmadminpw 
select 'begin dbms_streams_adm.add_table_propagation_rules(table_name => '''||OWNER||'.' || table_name || ''',' ||
        'streams_name => ''CAPTURE_MASTER'',' ||
        'source_queue_name => ''strmadmin.master_queue'',' ||
        'destination_queue_name => ''strmadmin.slave_queue'',' ||
        'include_dml => true,' || 'include_ddl => true,' ||
        'source_database => ''ora10g'',' || 'inclusion_rule => true,' ||
        'queue_to_queue => true); end; /'
  FROM TMP_STREAM_TABLES;
  
begin 
dbms_streams_adm.add_table_propagation_rules(
table_name => 'HURP.TEST',
streams_name => 'propagation_master',
source_queue_name => 'strmadmin.master_queue',
destination_queue_name => 'strmadmin.slave_queue@shrnc',
include_dml => true,
include_ddl => true,
source_database => 'ora10g',
inclusion_rule => true,
queue_to_queue => true);
end;
/  


BEGIN
DBMS_PROPAGATION_ADM.DROP_PROPAGATION(
propagation_name => 'CAPTURE_MASTER');
END;
/

#修改propagation休眠时间为0，表示实时传播LCR。 
begin 
dbms_aqadm.alter_propagation_schedule( 
queue_name => 'prod_queue', 
destination => 'shrnc', 
latency => 0); 
end; 
/
第 1 行出现错误:                           
ORA-24042: 不存在 QUEUE STRMADMIN.MASTER_QUEUE 及 DESTINATION SHRNC 的传播日程
ORA-06512: 在 "SYS.DBMS_SYS_ERROR", line 95                                                                                                                     
ORA-06512: 在 "SYS.DBMS_PRVTAQIP", line 957 
ORA-06512: 在 "SYS.DBMS_AQADM", line 910                                                                                                                          
ORA-06512: 在 line 2      
注：跳过以上报错，继续以下可以正常进行,后续查询相关资料，执行以下PL/SQL可以
begin                                       
dbms_aqadm.alter_propagation_schedule(      
queue_name => 'strmadmin.master_queue',                 
destination => 'shrnc',                    
destination_queue => 'strmadmin.slave_queue',        
latency => 0);                              
end;                                        
/                                           
创建应用进程:
以下是引用片段：
#以strmadmin身份，登录从数据库。 
connect strmadmin/strmadminpw 
从库命令窗口执行应用进程
select 'begin dbms_streams_adm.add_table_rules(table_name => '''||OWNER||'.' || table_name || ''',' ||
        'streams_type => ''apply'',' || 'streams_name => ''APPLY_SLAVE'',' ||
        'queue_name => ''strmadmin.slave_queue'',' ||
        'include_dml => true,' || 'include_ddl => true,' ||
        'include_tagged_lcr => false,' || 'source_database => ''ora10g'',' ||
        'inclusion_rule => true); end; /'
  FROM TMP_STREAM_TABLES ;

begin 
dbms_streams_adm.add_table_rules(
table_name => 'HURP.TEST',
streams_type =>'apply',
streams_name => 'APPLY_SLAVE',
queue_name => 'strmadmin.slave_queue',
include_dml => true,
include_ddl => true,
include_tagged_lcr => false,
source_database => 'ora10g',
inclusion_rule => true);
end;
/
Exec dbms_apply_adm.set_parameter(apply_name => 'APPLY_SLAVE',parameter => 'disable_on_error',value => 'n');
启停STREAM：
启动STREAM
以下是引用片段：
#以strmadmin身份，登录从数据库。 
connect strmadmin/strmadminpw
#启动Apply进程 
begin 
dbms_apply_adm.start_apply( 
apply_name => 'APPLY_SLAVE'); 
end; 
/ 
#以strmadmin身份，登录主数据库。 
connect strmadmin/strmadminpw
#启动Capture进程 
begin 
dbms_capture_adm.start_capture( 
capture_name => 'capture_master'); 
end; 
/
exec dbms_propagation_adm.start_propagation(propagation_name =>'propagation_master');
停止STREAM
以下是引用片段：
#以strmadmin身份，登录从数据库。 
connect strmadmin/strmadminpw
#启动Apply进程 
begin 
dbms_apply_adm.stop_apply( 
apply_name => 'APPLY_SLAVE'); 
end; 
/ 

#以strmadmin身份，登录主数据库。 
connect strmadmin/strmadminpw
#启动Capture进程 
begin 
dbms_capture_adm.stop_capture( 
capture_name => 'CAPTURE_MASTER'); 
end; 
/
问题诊断:
SELECT apply_name,queue_name,apply_captured,status,error_number,error_message FROM dba_apply@shrnc;
SELECT local_transaction_id,apply_name,queue_name,error_number,error_message,error_creation_time FROM dba_apply_error@shrnc ORDER BY error_creation_time DESC;

--以 strmadmin身份，登录从数据库，执行如下语句：
--主库执行
SELECT capture_name,status,queue_name,start_scn,captured_scn,applied_scn FROM dba_capture;
SELECT propagation_name,status,queue_to_queue from dba_propagation;
SELECT capture_name,state,capture_message_number,capture_message_create_time FROM v$streams_capture;


select sid,capture_name,startup_time,CAPTURE_TIME,CAPTURE_MESSAGE_NUMBER,state from v$streams_capture;


select sid,state,APPLY_NAME,TOTAL_RECEIVED,TOTAL_IGNORED,TOTAL_ERRORS from V$STREAMS_APPLY_COORDINATOR;


select streams_name,streams_type,CUMULATIVE_MESSAGE_COUNT,TOTAL_MESSAGE_COUNT from v$streams_transaction;
select PROPAGATION_NAME,SOURCE_QUEUE_OWNER,SOURCE_QUEUE_NAME,DESTINATION_QUEUE_OWNER,DESTINATION_QUEUE_NAME,DESTINATION_DBLINK,NEGATIVE_RULE_SET_OWNER,QUEUE_TO_QUEUE from dba_propagation;


批量插入脚本：
create or replace procedure batch_insert is 
v_sql02 varchar2(3999);
v_value number;
begin
for i in 1..10000 loop 
v_value:=i;
v_sql02:='insert into hurp.test values(:i,''a''||:i)';
execute immediate v_sql02 using v_value,v_value;   
commit;                                            
end loop;
end;
/




重新配置需要执行下列操作
从库执行：
要清除 Stream配置信息，需要先停止Stream进程。
#以strmadmin身份，登录主数据库。
--主库清理捕获进程
exec dbms_capture_adm.stop_capture(capture_name => 'CAPTURE_MASTER');
exec dbms_capture_adm.drop_capture(capture_name => 'CAPTURE_MASTER');


--主库清理分发进程
exec dbms_propagation_adm.stop_propagation(propagation_name =>  'propagation_master',force=>true);
exec dbms_propagation_adm.drop_propagation(propagation_name => 'propagation_master');


--主库清理队列
exec dbms_streams_adm.remove_queue(queue_name=>'master_queue');


--主库清理配置
exec dbms_streams_adm.remove_streams_configuration();
--从库停止应用进程
exec dbms_apply_adm.stop_apply(apply_name => 'APPLY_SLAVE');


--清理错误信息
exec dbms_apply_adm.delete_all_errors(apply_name => 'APPLY_SLAVE');


--从库清理应用进程
exec dbms_apply_adm.drop_apply(apply_name => 'APPLY_SLAVE');


--从库清理队列
exec dbms_streams_adm.remove_queue(queue_name=>'slave_queue');


--清除配置
exec dbms_streams_adm.remove_streams_configuration();
--各个库执行
truncate table sys.STREAMS$_APPLY_MILESTONE;
truncate table sys.STREAMS$_APPLY_PROCESS;
truncate table sys.STREAMS$_APPLY_PROGRESS;
truncate table sys.STREAMS$_APPLY_SPILL_MESSAGES;
truncate table sys.STREAMS$_APPLY_SPILL_MSGS_PART;
truncate table sys.STREAMS$_APPLY_SPILL_TXN;
truncate table sys.STREAMS$_APPLY_SPILL_TXN_LIST;
truncate table sys.STREAMS$_CAPTURE_PROCESS;
truncate table sys.STREAMS$_DEF_PROC;
truncate table sys.STREAMS$_DEST_OBJS;
truncate table sys.STREAMS$_DEST_OBJ_COLS;
truncate table sys.STREAMS$_EXTRA_ATTRS;
truncate table sys.STREAMS$_INTERNAL_TRANSFORM;
truncate table sys.STREAMS$_KEY_COLUMNS;
truncate table sys.STREAMS$_MESSAGE_CONSUMERS;
truncate table sys.STREAMS$_MESSAGE_RULES;
truncate table sys.STREAMS$_PREPARE_DDL;
truncate table sys.STREAMS$_PREPARE_OBJECT;
truncate table sys.STREAMS$_PRIVILEGED_USER;
truncate table sys.STREAMS$_PROCESS_PARAMS;
truncate table sys.STREAMS$_PROPAGATION_PROCESS;
truncate table sys.STREAMS$_RULES;
truncate table system.LOGMNRC_DBNAME_UID_MAP;
truncate table system.LOGMNRC_GSII;
truncate table system.LOGMNRC_GTCS;
truncate table system.LOGMNRC_GTLO;
truncate table system.LOGMNRP_CTAS_PART_MAP;
truncate table system.LOGMNRT_MDDL$;
truncate table system.LOGMNR_AGE_SPILL$;
truncate table system.LOGMNR_ATTRCOL$;
truncate table system.LOGMNR_ATTRIBUTE$;
truncate table system.LOGMNR_CCOL$;
truncate table system.LOGMNR_CDEF$;
truncate table system.LOGMNR_COL$;
truncate table system.LOGMNR_COLTYPE$;
truncate table system.LOGMNR_DICTIONARY$;
truncate table system.LOGMNR_DICTSTATE$;
truncate table system.LOGMNR_ERROR$;
truncate table system.LOGMNR_FILTER$;
truncate table system.LOGMNR_HEADER1$;
truncate table system.LOGMNR_HEADER2$;
truncate table system.LOGMNR_ICOL$;
truncate table system.LOGMNR_IND$;
truncate table system.LOGMNR_INDCOMPART$;
truncate table system.LOGMNR_INDPART$;
truncate table system.LOGMNR_INDSUBPART$;
truncate table system.LOGMNR_LOB$;
truncate table system.LOGMNR_LOBFRAG$;
truncate table system.LOGMNR_LOG$;
truncate table system.LOGMNR_OBJ$;
truncate table system.LOGMNR_PARAMETER$;
truncate table system.LOGMNR_PROCESSED_LOG$;
truncate table system.LOGMNR_RESTART_CKPT$;
truncate table system.LOGMNR_RESTART_CKPT_TXINFO$;
truncate table system.LOGMNR_SESSION$;
truncate table system.LOGMNR_SESSION_EVOLVE$;
truncate table system.LOGMNR_SPILL$;
truncate table system.LOGMNR_TAB$;
truncate table system.LOGMNR_TABCOMPART$;
truncate table system.LOGMNR_TABPART$;
truncate table system.LOGMNR_TABSUBPART$;
truncate table system.LOGMNR_TS$;
truncate table system.LOGMNR_TYPE$;
truncate table system.LOGMNR_UID$;
truncate table system.LOGMNR_USER$;