之前学习和联系stream环境时，都用的是DBMS_STREAMS_ADM包来配置的，在创建捕获进程后，每次要修改捕获规则时，都会自动增加新的捕获规则，并且新的条件不能生效，必须要删除原有的规则才可以。并且自动生成的规则名不易看懂，管理起来比较麻烦，认真学习了三思的大作 《 一步一步学Streams》后，尝试用手动建立规则的方式来搭建流环境，感觉比自动生成规则的方式方便很多，以下是实验步骤：
删除strmadmin 用户，为了干净的删除stream的规则等对象:
drop user strmadmin cascade;
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;
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;
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;
begin
dbms_streams_auth.grant_admin_privilege(grantee =>'strmadmin',grant_privileges => true);
end;
--建立数据库链
切换回source数据库，以streams的操作用户stradmin连接 ,创建队列
切换回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连接 ,创建队列
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();
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;
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
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;
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;
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;
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;
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;
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.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;
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;
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
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;
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.在捕获级别修改规则，即设置某些规则不捕获。
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;
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' )
(: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
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
---------- ----------
通过以上实验可以证明，在传播进程中设置条件也可以控制到那些表不传输。
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
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表不传。
实验需求，取消在传播进程规则中设置条件，修改为在捕获进程中设置条件。设置F6,F7表不传。
源库关闭传播进程。
SQL> exec dbms_propagation_adm.stop_propagation('PRI_TO_KLIR');
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_propagation_adm.start_propagation('PRI_TO_KLIR');
关闭捕获进程
SQL> exec dbms_capture_adm.stop_capture('LIRHZ_CAPTURE');
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
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');
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
----------
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
SQL> create table test02.f8 (id number,name varchar2(10));
Table created
目标库能看到f8表已经传输。
SQL> select * from test02.f8;
ID NAME
---------- ----------
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
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
本次实验参考了三思的一步步配置流系列文件，再次表示感谢！http://www.5ienet.com/note/html/streams/oracle10gr2_streams_single_source_rep_thr.shtml