2013年（350）
分类： Oracle
2013-04-27 09:41:59
一般有两种方式配置，一种是通过DBMS_CAPTURE_ADM、DBMS_PROPAGATION_ADM、DBMS_APPLY_ADM几个包分别创建捕获、传播和应用进程，相对来说更加灵活，但是操作比较复杂，另外一种就是直接借助DBMS_STREAMS_ADM包创建，这种方式相对比较简单，这里我们演示通过这种方式创建。
另外，本章中使用的环境延续了前面章节，初始化环境部分不再重复，如果你没有浏览本系列的其它文章，直接阅读本篇过程中无法理解部分数据来源，建议首先阅读前面部分相关章节。
下列操作如非特别说明，均是以strmadmin身份执行。
１、 Jssweb 端配置propagation进程JSSWEB> BEGIN
2 DBMS_STREAMS_ADM.ADD_SCHEMA_PROPAGATION_RULES(
3 schema_name => ¨member¨,
4 streams_name => ¨web_to_str¨,
5 source_queue_name => ¨strmadmin.streams_queue¨,
6 destination_queue_name => ¨strmadmin.streams_queue@jssstr.jss.cn¨,
7 include_dml => true,
8 include_ddl => true,
9 source_database => ¨jssweb.jss.cn¨,
10 inclusion_rule => true,
11 queue_to_queue => true);
12 END;
13 /
PL/SQL procedure successfully completed.JSSWEB> BEGIN
2 DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
3 schema_name => ¨member¨,
4 streams_type => ¨capture¨,
5 streams_name => ¨jssweb_capture¨,
6 queue_name => ¨strmadmin.streams_queue¨,
7 include_dml => true,
8 include_ddl => true,
9 inclusion_rule => true);
10 END;
11 /
PL/SQL procedure successfully completed.因为jssweb要复制到jssstr端的usr_user表在jssstr数据库已经存在(虽然改了名叫users)，因此我们需要设置初始scn，以让知道从何时开始的修改能够应用到目标端。
JSSSTR> DECLARE
2 iscn NUMBER;
3 BEGIN
4 iscn := DBMS_FLASHBACK.GET_SYSTEM_CHANGE_NUMBER();
5 DBMS_APPLY_ADM.SET_TABLE_INSTANTIATION_SCN@jssstr.jss.cn(
6 source_object_name => ¨member.usr_user¨,
7 source_database_name => ¨jssweb.jss.cn¨,
8 instantiation_scn => iscn);
9 END;
10 /
PL/SQL procedure successfully completed.执行本步的前提是jssweb和jssstr中member.usr_user两表完全相同，并且在执行该步的过程中，两端都不会对该表进行操作，不然可能导致后期修改和应用不一致。
４、 Jssbak 端从jssweb中导入对象Jssbak 中的schema:member中没有任何数据，因此在真正复制前，首先需要进行初始化，这里我们选择通过数据泵的方式从jssweb初始化数据到jssbak。
需要复制的表有三个：dt_tmp、dt_tbl1和dt_tbl2
JSSBAK> SET SERVEROUTPUT ON
DECLARE
h1 NUMBER; -- Pump job handle
sscn NUMBER; -- Variable to hold current source SCN
job_state VARCHAR2(30); -- To keep track of job state
js ku$_JobStatus; -- The job status from GET_STATUS
sts ku$_Status; -- The status object returned by GET_STATUS
JSSBAK> 2 3 4 5 6 7 job_not_exist exception;
8 pragma exception_init(job_not_exist, -31626);
9 BEGIN
10 -- Create a (user-named) Data Pump job to do a table-level import.
11 h1 := DBMS_DATAPUMP.OPEN(
12 peration => ¨IMPORT¨,
13 job_mode => ¨TABLE¨,
14 remote_link => ¨jssweb.jss.cn¨,
15 job_name => ¨dp_sync1¨);
16 -- A metadata filter is used to specify the schema that owns the tables
17 -- that will be imported.
18 DBMS_DATAPUMP.METADATA_FILTER(
19 handle => h1,
20 name => ¨SCHEMA_EXPR¨,
21 value => ¨=¨¨MEMBER¨¨¨);
22 -- A metadata filter is used to specify the tables that will be imported.
23 DBMS_DATAPUMP.METADATA_FILTER(
24 handle => h1,
25 name => ¨NAME_EXPR¨,
26 value => ¨IN(¨¨DT_TBL1¨¨, ¨¨DT_TBL2¨¨, ¨¨DT_TMP¨¨)¨);
27 -- Get the current SCN of the source database, and set the FLASHBACK_SCN
28 -- parameter to this value to ensure consistency between all of the
29 -- objects included in the import.
30 sscn := DBMS_FLASHBACK.GET_SYSTEM_CHANGE_NUMBER@jssweb.jss.cn();
31 DBMS_DATAPUMP.SET_PARAMETER(
32 handle => h1,
33 name => ¨FLASHBACK_SCN¨,
34 value => sscn);
35 -- Start the job.
36 DBMS_DATAPUMP.START_JOB(h1);
37 -- The import job should be running. In the following loop, the job
38 -- is monitored until it completes.
39 job_state := ¨UNDEFINED¨;
40 BEGIN
41 WHILE (job_state != ¨COMPLETED¨) AND (job_state != ¨STOPPED¨) LOOP
42 sts:=DBMS_DATAPUMP.GET_STATUS(
43 handle => h1,
44 mask => DBMS_DATAPUMP.KU$_STATUS_JOB_ERROR +
45 DBMS_DATAPUMP.KU$_STATUS_JOB_STATUS +
46 DBMS_DATAPUMP.KU$_STATUS_WIP,
47 timeout => -1);
48 js := sts.job_status;
49 DBMS_LOCK.SLEEP(10);
50 job_state := js.state;
51 END LOOP;
52 -- Gets an exception when job no longer exists
53 EXCEPTION WHEN job_not_exist THEN
54 DBMS_OUTPUT.PUT_LINE(¨Data Pump job has completed¨);
55 DBMS_OUTPUT.PUT_LINE(¨Instantiation SCN: ¨ ||sscn);
56 END;
57 END;
58 /
Data Pump job has completed
Instantiation SCN: 579689
PL/SQL procedure successfully completed.JSSBAK> BEGIN
2 DBMS_STREAMS_ADM.ADD_TABLE_RULES(
3 table_name => ¨member.dt_tmp¨,
4 streams_type => ¨apply¨,
5 streams_name => ¨jssbak_apply¨,
6 queue_name => ¨strmadmin.streams_queue¨,
7 include_dml => true,
8 include_ddl => true,
9 source_database => ¨jssweb.jss.cn¨,
10 inclusion_rule => true);
11
12 DBMS_STREAMS_ADM.ADD_TABLE_RULES(
13 table_name => ¨member.dt_tbl1¨,
14 streams_type => ¨apply¨,
15 streams_name => ¨jssbak_apply¨,
16 queue_name => ¨strmadmin.streams_queue¨,
17 include_dml => true,
18 include_ddl => true,
19 source_database => ¨jssweb.jss.cn¨,
20 inclusion_rule => true);
21
22 DBMS_STREAMS_ADM.ADD_TABLE_RULES(
23 table_name => ¨member.dt_tbl2¨,
24 streams_type => ¨apply¨,
25 streams_name => ¨jssbak_apply¨,
26 queue_name => ¨strmadmin.streams_queue¨,
27 include_dml => true,
28 include_ddl => true,
29 source_database => ¨jssweb.jss.cn¨,
30 inclusion_rule => true);
31 END;
32 /
PL/SQL procedure successfully completed.JSSBAK> exec DBMS_APPLY_ADM.START_APPLY(¨jssbak_apply¨);
PL/SQL procedure successfully completed.传播dt_tmp,dt_tbl1,dt_tbl2三表的修改到jssbak端：
JSSSTR> BEGIN
2 DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(
3 table_name => ¨member.dt_tmp¨,
4 streams_name => ¨str_to_bak¨,
5 source_queue_name => ¨strmadmin.streams_queue¨,
6 destination_queue_name => ¨strmadmin.streams_queue@jssbak.jss.cn¨,
7 include_dml => true,
8 include_ddl => true,
9 source_database => ¨jssweb.jss.cn¨,
10 inclusion_rule => true,
11 queue_to_queue => true);
12
13 DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(
14 table_name => ¨member.dt_tbl1¨,
15 streams_name => ¨str_to_bak¨,
16 source_queue_name => ¨strmadmin.streams_queue¨,
17 destination_queue_name => ¨strmadmin.streams_queue@jssbak.jss.cn¨,
18 include_dml => true,
19 include_ddl => true,
20 source_database => ¨jssweb.jss.cn¨,
21 inclusion_rule => true);
22
23 DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(
24 table_name => ¨member.dt_tbl2¨,
25 streams_name => ¨str_to_bak¨,
26 source_queue_name => ¨strmadmin.streams_queue¨,
27 destination_queue_name => ¨strmadmin.streams_queue@jssbak.jss.cn¨,
28 include_dml => true,
29 include_ddl => true,
30 source_database => ¨jssweb.jss.cn¨,
31 inclusion_rule => true);
32 END;
33 /
PL/SQL procedure successfully completed.JSSSTR> CREATE OR REPLACE FUNCTION member.to_users_trans_dml(
2 p_in_data in ANYDATA)
3 RETURN ANYDATA IS out_data SYS.LCR$_ROW_RECORD;
4 tc pls_integer;
5 BEGIN
6 -- Typecast AnyData to LCR$_ROW_RECORD
7 tc := p_in_data.GetObject(out_data);
8 IF out_data.GET_OBJECT_NAME() = ¨USR_USER¨
9 THEN
10 -- Transform. the in_data into the out_data
11 out_data.SET_OBJECT_NAME(¨USERS¨);
12 END IF;
13 -- Convert to AnyData
14 RETURN ANYDATA.ConvertObject(out_data);
15 END;
16 /
Function created.注意指定dml_rule：
JSSSTR> set serveroutput on
JSSSTR> DECLARE
2 to_users_rulename_dml VARCHAR2(30);
3 dummy_rule VARCHAR2(30);
4 BEGIN
5 DBMS_STREAMS_ADM.ADD_TABLE_RULES(
6 table_name => ¨member.usr_user¨,
7 streams_type => ¨apply¨,
8 streams_name => ¨jssstr_apply¨,
9 queue_name => ¨strmadmin.streams_queue¨,
10 include_dml => true,
11 include_ddl => false,
12 source_database => ¨jssweb.jss.cn¨,
13 dml_rule_name => to_users_rulename_dml,
14 ddl_rule_name => dummy_rule,
15 inclusion_rule => true);
16 DBMS_OUTPUT.PUT_LINE(¨Dml rule name: ¨|| to_users_rulename_dml);
17 END;
18 /
Dml rule name: "STRMADMIN"."USR_USER150"
PL/SQL procedure successfully completed.设置输出的dml_rule执行自定义的函数：
JSSSTR> BEGIN
2 DBMS_STREAMS_ADM.SET_RULE_TRANSFORM_FUNCTION(
3 rule_name => ¨STRMADMIN.USR_USER150¨,
4 transform_function => ¨member.to_users_trans_dml¨);
5 END;
6 /
PL/SQL procedure successfully completed.JSSSTR> exec dbms_apply_adm.start_apply(¨jssstr_apply¨);
PL/SQL procedure successfully completed.JSSWEB> EXEC DBMS_CAPTURE_ADM.START_CAPTURE(¨jssweb_capture¨);
PL/SQL procedure successfully completed.首先来测试jssweb到jssstr端的dml操作(不复制ddl操作，因此ddl部分不测试)：
JSSWEB> select * from member.usr_user;
no rows selected
JSSWEB> insert into member.usr_user values (10,¨ALLEN¨);
1 row created.
JSSWEB> commit;
Commit complete.
JSSWEB> alter system switch logfile;
System altered.
JSSWEB> select * from member.users@jssstr.jss.cn;
USER_ID USERNAME
---------- ------------------------------
10 ALLEN再来测试一下jssweb捕获，jssstr传播，jssbak应用的情况：
JSSWEB> select count(0) from member.dt_tmp;
COUNT(0)
----------
0
JSSWEB> insert into member.dt_tmp values (10);
1 row created.
JSSWEB> commit;
Commit complete.
JSSWEB> alter system switch logfile;
System altered.Jssweb 端没有直接连接jssbak的链，因此重新连接到jssbak端查看：
JSSBAK> select * from member.dt_tmp;
ID
----------
10再测试一下ddl应用的情况：
JSSWEB> desc member.dt_tbl2;
Name Null? Type
----------------------------------------- -------- ----------------------------
ID NUMBER
JSSWEB> alter table member.dt_tbl2 add vname varchar2(20);
Table altered.
JSSWEB> alter system switch logfile;
System altered.转到jssbak端查看：
JSSBAK> desc member.dt_tbl2;
Name Null? Type
----------------------------------------- -------- ----------------------------
ID NUMBER
VNAME VARCHAR2(20)竣工。
======================================
