手动建立规则方式创建stream环境

 

删除strmadmin 用户，为了干净的删除stream的规则等对象:
drop user strmadmin cascade;
 

源库建立strmadmin用户,并授权.
create user strmadmin identified by strmadmin default tablespace tbs_stream temporary tablespace temp;
grant connect,resource,dba,aq_administrator_role to strmadmin;

目标库建立strmadmin用户,并授权.
create user strmadmin identified by strmadmin default tablespace tbs_stream temporary tablespace temp;
grant connect,resource,dba,aq_administrator_role to strmadmin;

授权Stream管理用户 ：
begin
dbms_streams_auth.grant_admin_privilege(grantee =>'strmadmin',grant_privileges => true);
end;
  
 

 --建立数据库链
 切换回source数据库，以streams的操作用户stradmin连接  ,创建队列

 create database link klir connect to strmadmin identified by strmadmin using 'klir';
 exec dbms_streams_adm.set_up_queue(); 
 
 
 切换到target数据库，以streams的操作用户stradmin连接 ,创建队列

  create database link lirhz connect to strmadmin identified by strmadmin using 'lirhz';
  exec dbms_streams_adm.set_up_queue();

 

 

手动建立传播规则和传播进程：

先建立一个传播进程的规则集：
sour> begin
dbms_rule_adm.create_rule_set(
rule_set_name => 'strmadmin.propagation_rules',
evaluation_context => 'sys.streams$_evaluation_context');
end;

建立规则集后，会在（dba_rulesets、dba_rule_sets）表中建立propagation_rules规则。

创建传播的dml规则。
sour> begin
dbms_rule_adm.create_rule(
rule_name => 'strmadmin.member_pro_dml',
condition => ':dml.get_object_owner()=''TEST02'' AND ' ||
' :dml.is_null_tag() =''Y'' AND '||
' :dml.get_source_database_name()= ''LIRHZ'' ');
end;
   
写入表：  dba_rules

创建传播的ddl规则。
begin
   dbms_rule_adm.create_rule(
   rule_name => 'strmadmin.member_pro_ddl',
   condition => '(:ddl.get_object_owner()=''TEST02'' OR ' ||
   ' :ddl.get_base_table_owner() =''TEST02'') AND ' ||
   ' :ddl.is_null_tag() =''Y'' AND '||
   ' :ddl.get_source_database_name()= ''LIRHZ'' ');
end;
   
把dml、ddl规则添加到规则集中
   
begin
   dbms_rule_adm.add_rule(
   rule_name => 'strmadmin.member_pro_dml',
   rule_set_name => 'strmadmin.propagation_rules');
end;

把ddl规则添加到规则集合中 ,
begin
   dbms_rule_adm.add_rule(
   rule_name => 'strmadmin.member_pro_ddl',
   rule_set_name => 'strmadmin.propagation_rules');
end;

创建propagation进程。

begin
   dbms_propagation_adm.create_propagation(
   propagation_name => 'pri_to_klir',
   source_queue => 'strmadmin.streams_queue',
   destination_queue => 'strmadmin.streams_queue',
   destination_dblink => 'klir',
   rule_set_name => 'strmadmin.propagation_rules');
   end;

 

创建capture进程部分。
手动建立捕获规则和捕获进程：

创建规则集
sour> begin
   dbms_rule_adm.create_rule_set(
   rule_set_name => 'strmadmin.capture_rules',
   evaluation_context => 'sys.streams$_evaluation_context');
   end;

创建dml规则
begin
   dbms_rule_adm.create_rule(
   rule_name => 'strmadmin.member_cap_dml',
   condition => ':dml.get_object_owner()=''TEST02'' AND ' ||
   ' :dml.is_null_tag() =''Y'' AND '||
   ' :dml.get_source_database_name()= ''LIRHZ'' ');
   end;

创建ddl规则   
sour> begin
   dbms_rule_adm.create_rule(
   rule_name => 'strmadmin.member_cap_ddl',
   condition => '(:ddl.get_object_owner()=''TEST02'' OR ' ||
   ' :ddl.get_base_table_owner() =''TEST02'') AND ' ||
   ' :ddl.is_null_tag() =''Y'' AND '||
   ' :ddl.get_source_database_name()= ''LIRHZ'' ');
   end;

   
加入规则集。
sour> begin
   dbms_rule_adm.add_rule(
   rule_name => 'strmadmin.member_cap_dml',
   rule_set_name => 'strmadmin.capture_rules');
   end;
 

sour> begin
   dbms_rule_adm.add_rule(
   rule_name => 'strmadmin.member_cap_ddl',
   rule_set_name => 'strmadmin.capture_rules');
   end;

创建捕获进程
 begin
   dbms_capture_adm.create_capture(
   queue_name => 'strmadmin.streams_queue',
   capture_name => 'lirhz_capture',
   rule_set_name => 'strmadmin.capture_rules');   使用捕获的规则集。
   end;

 

源库建立的捕获传播规则

SQL>  select RULE_NAME,RULE_CONDITION from dba_rules where rule_owner='STRMADMIN';
 
RULE_NAME                      RULE_CONDITION
------------------------------ --------------------------------------------------------------------------------
MEMBER_CAP_DDL                 (:ddl.get_object_owner()='TEST02' OR  :ddl.get_base_table_owner() ='TEST02') AND
MEMBER_CAP_DML                 :dml.get_object_owner()='TEST02' AND  :dml.is_null_tag() ='Y' AND  :dml.get_sour
MEMBER_PRO_DDL                 (:ddl.get_object_owner()='TEST02' OR  :ddl.get_base_table_owner() ='TEST02') AND
MEMBER_PRO_DML                 :dml.get_object_owner()='TEST02' AND  :dml.is_null_tag() ='Y' AND  :dml.get_sour

 

 

源端准备实例SCN （如果是使用dbms_streams_adm包来配置就不需要此步骤，过程会自动进行配置。）。
begin
   dbms_capture_adm.prepare_schema_instantiation(
   schema_name => 'test02',
   supplemental_logging => 'keys');
   end;

目标库做应用scn。
declare
 iscn number;
 begin
 iscn:=dbms_flashback.get_system_change_number();
  dbms_apply_adm.set_schema_instantiation_scn@klir(
 source_schema_name => 'test02',
 source_database_name => 'lirhz',
 instantiation_scn => iscn,
 recursive  => true);
end;

目标库使用dbms_streams_adm。建立apply.建立后的规则由ORACLE自动命名。
begin
dbms_streams_adm.add_schema_rules(
schema_name => 'test02',
streams_type => 'apply',
streams_name => 'apply_standy',
queue_name => 'strmadmin.STREAMS_QUEUE',
include_dml => true,
include_ddl => true,
include_tagged_lcr => false,
source_database => 'lirhz',
inclusion_rule => true);
end;

设置异常规则。

启动应用进程和传播进程后，这样单向schema级的数据就能同步了。忽略.........

 

========================================================

我使用手动建立规则的方式去配置流，主要是考虑在实际应用中可以需要修改和维护同步规则，例如要设置源库的个别表不传到目标库，有2种方式：
1.在传播级别修改规则，即设置某些规则不传播。
2.在捕获级别修改规则，即设置某些规则不捕获。

原先的传播规则命令 
begin
   dbms_rule_adm.create_rule(
   rule_name => 'strmadmin.member_pro_ddl',
   condition => '(:ddl.get_object_owner()=''TEST02'' OR ' ||
   ' :ddl.get_base_table_owner() =''TEST02'') AND ' ||
   ' :ddl.is_null_tag() =''Y'' AND '||
   ' :ddl.get_source_database_name()= ''LIRHZ'' ');
end;

修改dml传播规则。增加F1,F5的不传条件。

sour> begin
dbms_rule_adm.alter_rule(
rule_name => 'strmadmin.member_pro_dml',
condition => ':dml.get_object_owner()=''TEST02'' AND ' ||
' :dml.is_null_tag() =''Y'' AND '||
' :dml.get_source_database_name()= ''LIRHZ'' and ' ||
' (:dml.get_object_name() <> ''F1'' and :dml.get_object_name() <> ''F5'' ) ');
end;

修改ddl传播规则。
begin
   dbms_rule_adm.alter_rule(
   rule_name => 'strmadmin.member_pro_ddl',
   condition => '(:ddl.get_object_owner()=''TEST02'' OR ' ||
   ' :ddl.get_base_table_owner() =''TEST02'') AND ' ||
   ' :ddl.is_null_tag() =''Y'' AND '||
   ' :ddl.get_source_database_name()= ''LIRHZ'' and ' ||
   ' (:dml.get_object_name() <> ''F1'' and :dml.get_object_name() <> ''F5'' ) ');
end;

实际上增加的条件在dba_rules 表的RULE_CONDITION字段中有显示，值如下。
(:ddl.get_object_owner()='TEST02' OR  :ddl.get_base_table_owner() ='TEST02') AND  :ddl.is_null_tag() ='Y' AND  :ddl.get_source_database_name()= 'LIRHZ' and  (:dml.get_object_name() <> 'F1' and :dml.get_object_name() <> 'F5' )
                                            :dml.get_object_owner()='TEST02' AND  :dml.is_null_tag() ='Y' AND  :dml.get_source_database_name()= 'LIRHZ' and  (:dml.get_object_name() <> 'F1' and :dml.get_object_name() <> 'F5' )

 

源库启动传播进程，
SQL> exec dbms_propagation_adm.start_propagation('PRI_TO_KLIR');
 
PL/SQL procedure successfully completed
 
源库的f1,f5分别插入记录。
SQL>  insert into test02.f1 values (33,'guangzhou');
1 row inserted
 
SQL>  insert into test02.f1 values (34,'guangzhou');
1 row inserted
 
SQL>  insert into test02.f5 values (34,'guangzhou');
1 row inserted
 
SQL>  insert into test02.f5 values (33,'guangzhou');
1 row inserted
 
SQL> commit;
Commit complete

源库可以看到新加的33，34记录，
SQL> select * from test02.f5 where id in (33,34);
 
        ID NAME
---------- ----------
        33 guangzhou
        34 guangzhou
 
SQL> select * from test02.f1 where id in (33,34);
 
        ID NAME
---------- ----------
        33 guangzhou
        34 guangzhou
        
       

转到目标库查询时，无对应的33,34记录
SQL> select * from test02.f5 where id in (33,34);
 
        ID NAME
---------- ----------
 
SQL> select * from test02.f1 where id in (33,34);
 
        ID NAME
---------- ----------
通过以上实验可以证明，在传播进程中设置条件也可以控制到那些表不传输。

但查看dba_capture视图可以看到几个scn号已经完全不一致了。因为捕获的日志不一定都能应用了。
SQL> select capture_name,START_SCN,STATUS,CAPTURED_SCN,APPLIED_SCN  from dba_capture;
 
CAPTURE_NAME                    START_SCN STATUS   CAPTURED_SCN APPLIED_SCN
------------------------------ ---------- -------- ------------ -----------
LIRHZ_CAPTURE                  2582441264 ENABLED    2582659272  2582659140

 

继续实验：
  实验需求，取消在传播进程规则中设置条件，修改为在捕获进程中设置条件。设置F6,F7表不传。

源库关闭传播进程。
SQL> exec dbms_propagation_adm.stop_propagation('PRI_TO_KLIR');

修改dml传播规则。重新删除F1,F5表的不传条件。
sour> begin
dbms_rule_adm.alter_rule(
rule_name => 'strmadmin.member_pro_dml',
condition => ':dml.get_object_owner()=''TEST02'' AND ' ||
' :dml.is_null_tag() =''Y'' AND '||
' :dml.get_source_database_name()= ''LIRHZ'' ');
end;

修改ddl传播规则。
begin
   dbms_rule_adm.alter_rule(
   rule_name => 'strmadmin.member_pro_ddl',
   condition => '(:ddl.get_object_owner()=''TEST02'' OR ' ||
   ' :ddl.get_base_table_owner() =''TEST02'') AND ' ||
   ' :ddl.is_null_tag() =''Y'' AND '||
   ' :ddl.get_source_database_name()= ''LIRHZ'' ');
end;

源库启动传播进程。
SQL> exec dbms_propagation_adm.start_propagation('PRI_TO_KLIR');

关闭捕获进程
SQL> exec dbms_capture_adm.stop_capture('LIRHZ_CAPTURE');

源库在f6,f7表中各插入2条记录。

SQL> insert into test02.f6 values (20);
1 row inserted
 
SQL> insert into test02.f6 values (21);
1 row inserted
 
SQL> insert into test02.f7 values (21);
1 row inserted
 
SQL> insert into test02.f7 values (20);
1 row inserted
 
SQL> commit;
Commit complete

修改源库的捕获规则
 
begin
   dbms_rule_adm.alter_rule(
   rule_name => 'strmadmin.member_cap_dml',
   condition => ':dml.get_object_owner()=''TEST02'' AND ' ||
   ' :dml.is_null_tag() =''Y'' AND '||
   ' :dml.get_source_database_name()= ''LIRHZ'' and ' ||
   ' (:dml.get_object_name() <> ''F6'' and :dml.get_object_name() <> ''F7'' ) ');
   end;
  

sour> begin
   dbms_rule_adm.alter_rule(
   rule_name => 'strmadmin.member_cap_ddl',
   condition => '(:ddl.get_object_owner()=''TEST02'' OR ' ||
   ' :ddl.get_base_table_owner() =''TEST02'') AND ' ||
   ' :ddl.is_null_tag() =''Y'' AND '||
   ' :ddl.get_source_database_name()= ''LIRHZ'' and ' ||
   ' (:dml.get_object_name() <> ''F6'' and :dml.get_object_name() <> ''F7'' ) ');
   end;

启动捕获进程：
SQL> exec dbms_capture_adm.start_capture('LIRHZ_CAPTURE');

查看目标库的f6.f7表,修改捕获规则前源库插入的2个记录不会再传到目标库了。
SQL> select * from test02.f6 where id in (20,21);
 
        ID
----------
 
SQL> select * from test02.f7 where id in (20,21);
 
        ID
----------
 

源库创建f8表，
SQL> create table test02.f8 (id number,name varchar2(10));
 
Table created

目标库能看到f8表已经传输。

SQL> select * from test02.f8;
 
        ID NAME
---------- ----------

可以看到dba_capture字典中记录的捕获和应用scn是一样的了。
SQL> select capture_name,START_SCN,STATUS,CAPTURED_SCN,APPLIED_SCN  from dba_capture;
 
CAPTURE_NAME                    START_SCN STATUS   CAPTURED_SCN APPLIED_SCN
------------------------------ ---------- -------- ------------ -----------
LIRHZ_CAPTURE                  2582441264 ENABLED    2582752740  2582752740

本文转自 gjm008 51CTO博客，原文链接：http://blog.51cto.com/gaoshan/418540，如需转载请自行联系原作者