Oracle 10g 流复制（Streams Replication）配置

Oracle 流是一种数据共享的通用机制，可以被用于许多处理的基础，包括消息、复制和数据仓库的 ETL 过程。它是高级队列、LogMinor、作业调度等已存在技术的扩展。这里做一个简单的复制示例来说明其用法。试验环境：

windows server 2003 sp1；10g 10.1.0.2
源库：SID：db1 IP：10.1.8.201
目的库：SID：db2 IP：10.1.9.49

源、目的库的 tnsnames.ora 配置
DB1 =
(DESCRIPTION =
(ADDRESS = (PROTOCOL = TCP)(HOST = 10.1.8.201)(PORT = 1521))
(CONNECT_DATA =
(SERVICE_NAME = db1)
(SERVER = DEDICATED)
)
)

DB2 =
(DESCRIPTION =
(ADDRESS_LIST =
(ADDRESS = (PROTOCOL = TCP)(HOST = 10.1.9.49)(PORT = 1521))
)
(CONNECT_DATA =
(SERVICE_NAME = db2)
(SERVER = DEDICATED)
)
)



1. 基本体系结构

流处理分为捕捉、传输、应用三个主要进程。

捕捉进程是一个可选的后台进程。它从重做日志中捕获 DDL 和 DML 的变化，并且把它们封装成逻辑改变记录（LCRs）。缺省的捕捉机制可以立即获得用户定义的事件。
传输进程把 LCRs 存储到 SYS.AnyData 数据类型的队列中。LCRs 在不同数据库中的源和目的传输区域之间传播。传播使用工作队列来调度。
应用进程是一个可选的后台进程。它调出 LCRs 直接应用，或者为用户自定义消息作为参数传递给用户自定义包。

捕获和应用进程可以被用于表、模式、数据库等级别，各级别的规则决定了其不同的行为。

2. 实例设置

CONNsys/password@DB1ASSYSDBA
createpfilefromspfile;
--编辑initsid.ora文件，增加以下参数：
--AQ_TM_PROCESSES=1;
--GLOBAL_NAMES=TRUE;

SHUTDOWNIMMEDIATE;
createspfilefrompfile;
STARTUPmount;
alterdatabasearchivelog;--改为归档模式
alterdatabaseopen;

3. 流管理员设置

CONNsys/password@DB1ASSYSDBA

CREATEUSERstrmadminIDENTIFIEDBYstrmadminDEFAULTTABLESPACEusersQUOTAUNLIMITEDONusers;

GRANTDBA,CONNECT,RESOURCE,SELECT_CATALOG_ROLETOstrmadmin;

GRANTEXECUTEONDBMS_AQADMTOstrmadmin;
GRANTEXECUTEONDBMS_CAPTURE_ADMTOstrmadmin;
GRANTEXECUTEONDBMS_PROPAGATION_ADMTOstrmadmin;
GRANTEXECUTEONDBMS_STREAMS_ADMTOstrmadmin;
GRANTEXECUTEONDBMS_APPLY_ADMTOstrmadmin;
GRANTEXECUTEONDBMS_FLASHBACKTOstrmadmin;

GRANTALLONscott.deptTOstrmadmin;

BEGIN
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege=>DBMS_RULE_ADM.CREATE_RULE_SET_OBJ,
grantee=>'strmadmin',
grant_option=>FALSE);
END;
/

BEGIN
DBMS_RULE_ADM.GRANT_SYSTEM_PRIVILEGE(
privilege=>DBMS_RULE_ADM.CREATE_RULE_OBJ,
grantee=>'strmadmin',
grant_option=>FALSE);
END;
/

CONNECTstrmadmin/strmadmin@DB1
EXECDBMS_STREAMS_ADM.SET_UP_QUEUE();--建立流队列

CREATEDATABASELINKdb2CONNECTTOstrmadminIDENTIFIEDBYstrmadminUSING'DB2';--建立数据库连接

在目的库（DB2）重复上面2、3步骤。

4. LogMinor 表空间设置

CONNsys/password@DB1ASSYSDBA

CREATETABLESPACElogmnr_tsDATAFILE'E:ORACLEPRODUCT10.1.0ORADATADB1logmnr01.dbf'SIZE25MREUSEAUTOEXTENDONMAXSIZEUNLIMITED;

EXECUTEDBMS_LOGMNR_D.SET_TABLESPACE('logmnr_ts');

5. SUPPLEMENTAL日志

CONNsys/password@DB1ASSYSDBA
ALTERTABLEscott.deptADDSUPPLEMENTALLOGGROUPlog_group_dept_pk(deptno)ALWAYS;

6. 配置传播进程

CONNECTstrmadmin/strmadmin@DB1
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(
table_name=>'scott.dept',
streams_name=>'db1_to_db2',
source_queue_name=>'strmadmin.streams_queue',
destination_queue_name=>'strmadmin.streams_queue@db2',
include_dml=>true,
include_ddl=>true,
source_database=>'db1');
END;
/

--检查传播进程作业
SELECTjob,TO_CHAR(last_date,'DD-Mon-YYYYHH24:MI:SS')last_date,TO_CHAR(next_date,'DD-Mon-YYYYHH24:MI:SS')next_date,whatFROMdba_jobs;

7. 配置捕捉进程

CONNECTstrmadmin/strmadmin@DB1
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name=>'scott.dept',
streams_type=>'capture',
streams_name=>'capture_simp',
queue_name=>'strmadmin.streams_queue',
include_dml=>true,
include_ddl=>true);
END;
/

8. 配置初始 SCN
在应用进程工作前必须在目的表中配置源表的 SCN。如果目的表已经存在，可用 exp/imp 元数据完成。

exp userid=scott/tiger@db1 FILE=dept_instant.dmp TABLES=dept OBJECT_CONSISTENT=y ROWS=n
imp userid=scott/tiger@db2 FILE=dept_instant.dmp IGNORE=y COMMIT=y LOG=import.log STREAMS_INSTANTIATION=y

因为在迁移元数据时 SUPPLEMENTAL 日志也被迁移。因为捕捉进程还没有启动，所以可以删除这些日志：

CONNsys/password@DB2ASSYSDBA
ALTERTABLEscott.deptDROPSUPPLEMENTALLOGGROUPlog_group_dept_pk;

用 DBMS_APPLY_ADM 包设置 SCN

CONNECTstrmadmin/strmadmin@db1
DECLARE
v_scnNUMBER;
BEGIN
v_scn:=DBMS_FLASHBACK.GET_SYSTEM_CHANGE_NUMBER();
DBMS_APPLY_ADM.SET_TABLE_INSTANTIATION_SCN@DB2(
source_object_name=>'scott.dept',
source_database_name=>'db1',
instantiation_scn=>v_scn);
END;
/

9. 配置应用进程

CONNECTstrmadmin/strmadmin@DB2
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name=>'scott.dept',
streams_type=>'apply',
streams_name=>'apply_simp',
queue_name=>'strmadmin.streams_queue',
include_dml=>true,
include_ddl=>true,
source_database=>'db1');
END;
/

10. 启动应用进程

CONNECTstrmadmin/strmadmin@DB2
BEGIN
DBMS_APPLY_ADM.SET_PARAMETER(
apply_name=>'apply_simp',
parameter=>'disable_on_error',
value=>'n');

DBMS_APPLY_ADM.START_APPLY(
apply_name=>'apply_simp');
END;
/

11. 启动捕捉进程

CONNECTstrmadmin/strmadmin@DB1
BEGIN
DBMS_CAPTURE_ADM.START_CAPTURE(
capture_name=>'capture_simp');
END;
/

12. 测试

--测试DML

CONNECTscott/tiger@db1
INSERTINTOdept(deptno,dname,loc)VALUES(99,'TestDept','UK');
COMMIT;

SELECT*FROMdept;

DEPTNODNAMELOC
-------------------------------------
10ACCOUNTINGNEWYORK
20RESEARCHDALLAS
30SALESCHICAGO
40OPERATIONSBOSTON
99TestDeptUK

5rowsselected.

CONNECTscott/tiger@db2
SELECT*FROMdept;

DEPTNODNAMELOC
-------------------------------------
10ACCOUNTINGNEWYORK
20RESEARCHDALLAS
30SALESCHICAGO
40OPERATIONSBOSTON
99TestDeptUK

5rowsselected.

--测试DDL

CONNECTscott/tiger@db1
ALTERTABLEdeptADD(new_colNUMBER(10))
/
DESCdept

NameNull?Type
--------------------------------------------------
DEPTNONOTNULLNUMBER(2)
DNAMEVARCHAR2(14)
LOCVARCHAR2(13)
NEW_COLNUMBER(10)

CONNECTscott/tiger@db2
DESCdept

NameNull?Type
--------------------------------------------------
DEPTNONOTNULLNUMBER(2)
DNAMEVARCHAR2(14)
LOCVARCHAR2(13)
NEW_COLNUMBER(10)

--可以用下列语句查看流的内容
CONNECTstrmadmin/strmadmin@DB1

SELECTs.user_data.getTypeName()FROMstreams_queue_tables;

SETSERVEROUTPUTON
DECLARE
v_anydataSYS.ANYDATA;
v_lcrSYS.LCR$_ROW_RECORD;
v_row_listSYS.LCR$_ROW_LIST;
v_resultPLS_INTEGER;
BEGIN

SELECTuser_data
INTOv_anydata
FROMstrmadmin.streams_queue_table
WHERErownum<2;

v_result:=ANYDATA.GetObject(
self=>v_anydata,
obj=>v_lcr);

DBMS_OUTPUT.PUT_LINE('CommandType:'||v_lcr.Get_Command_Type);
DBMS_OUTPUT.PUT_LINE('ObjectOwner:'||v_lcr.Get_Object_Owner);
DBMS_OUTPUT.PUT_LINE('ObjectName:'||v_lcr.Get_Object_Name);
DBMS_OUTPUT.PUT_LINE('SourceDatabaseName:'||v_lcr.Get_Source_Database_Name);
END;
/

13. 清理

--可以使用下面的语句识别并删除所有规则
conn/assysdba

BEGIN
FORcur_recIN(SELECTrule_owner,rule_nameFROMdba_rules)LOOP
DBMS_RULE_ADM.DROP_RULE(
rule_name=>cur_rec.rule_owner||'.'||cur_rec.rule_name,
force=>TRUE);
ENDLOOP;
END;
/

--识别、停止并删除所有捕捉进程和应用进程
conn/assysdba

BEGIN
FORcur_recIN(SELECTcapture_nameFROMdba_capture)LOOP
DBMS_CAPTURE_ADM.STOP_CAPTURE(
capture_name=>cur_rec.capture_name);
DBMS_CAPTURE_ADM.DROP_CAPTURE(
capture_name=>cur_rec.capture_name);
ENDLOOP;

FORcur_recIN(SELECTapply_nameFROMdba_apply)LOOP
DBMS_APPLY_ADM.STOP_APPLY(
apply_name=>cur_rec.apply_name);
DBMS_APPLY_ADM.DROP_APPLY(
apply_name=>cur_rec.apply_name);
ENDLOOP;
END;
/

--使用下面的语句彻底删除相关对象的流信息
BEGIN
DBMS_STREAMS_ADM.PURGE_SOURCE_CATALOG(
source_database=>'db1',
source_object_name=>'scott.dept',
source_object_type=>'TABLE');
END;
/

相关信息：
Oracle9i Streams Release 2 (9.2)
Monitoring a Streams Environment
Oracle9i Supplied PL/SQL Packages and Types Reference Release 2 (9.2)