一步一步学Streams -- 第三部分(21)高级配置之单源多向复制环境(2)

首页：三思笔记 > 一步一步学系列 > 一步一步学Streams

第三部分(21)高级配置之单源多向复制环境(2)

[君三思] 2009-3-30

四、简单方式配置复制环境

　　一般有两种方式配置，一种是通过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 /

２、 Jssweb 端配置capture进程

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 /

３、 Jssstr 端对已存在的表设置实例scn

因为jssweb要复制到jssstr端的usr_user表在jssstr数据库已经存在(虽然改了名叫users)，因此我们需要设置初始scn，以让streams知道从何时开始的修改能够应用到目标端。

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 /

执行本步的前提是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; -- Data 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 operation => ¨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

５、 JSSBAK 端配置apply进程

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);

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);

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 /

６、 JSSBAK 端启动apply进程

JSSBAK> exec DBMS_APPLY_ADM.START_APPLY(¨jssbak_apply¨);

７、 JSSSTR 端配置propagation进程

传播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);

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);

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 /

８、 Jssstr 端设置名称转换函数

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 /

９、 Jssstr 端设置apply进程

注意指定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"

设置输出的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 /

１０、 Jssstr 端启动apply进程

JSSSTR> exec dbms_apply_adm.start_apply(¨jssstr_apply¨);

１１、 Jssweb 端启动capture进程

JSSWEB> EXEC DBMS_CAPTURE_ADM.START_CAPTURE(¨jssweb_capture¨);

１２、测试

首先来测试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

---------- ------------------------------

再来测试一下jssweb捕获，jssstr传播，jssbak应用的情况：

JSSWEB> select count(0) from member.dt_tmp;

COUNT(0)

----------

JSSWEB> insert into member.dt_tmp values (10);

1 row created.

JSSWEB> commit;

Commit complete.

JSSWEB> alter system switch logfile;

Jssweb 端没有直接连接jssbak的数据库链，因此重新连接到jssbak端查看：

JSSBAK> select * from member.dt_tmp;

----------

再测试一下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;

转到jssbak端查看：

JSSBAK> desc member.dt_tbl2;

Name Null? Type

----------------------------------------- -------- ----------------------------

ID NUMBER

竣工。

上一条：第三部分(20)高级配置之单源多向复制环境(1)
下一条：第三部分(22)高级配置之单源多向复制环境(3)
[返回导航]

四、 简单方式配置复制环境

１、 Jssweb 端配置propagation进程

２、 Jssweb 端配置capture进程

３、 Jssstr 端对已存在的表设置实例scn

４、 Jssbak 端从jssweb中导入对象

５、 JSSBAK 端配置apply进程

６、 JSSBAK 端启动apply进程

７、 JSSSTR 端配置propagation进程

８、 Jssstr 端设置名称转换函数

９、 Jssstr 端设置apply进程

１０、 Jssstr 端启动apply进程

１１、 Jssweb 端启动capture进程

１２、 测试

四、简单方式配置复制环境

１２、测试