Oracle 10g 流复制（Streams Replication）配置
2016-12-29 14:11
381 查看
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. 实例设置
CONN sys/password@DB1 AS SYSDBA
create pfile from spfile;
-- 编辑 initsid.ora 文件，增加以下参数：
-- AQ_TM_PROCESSES=1;
-- GLOBAL_NAMES=TRUE;
SHUTDOWN IMMEDIATE;
create spfile from pfile;
STARTUP mount;
alter database archivelog; -- 改为归档模式
alter database open;
3. 流管理员设置
CONN sys/password@DB1 AS SYSDBA
CREATE USER strmadmin IDENTIFIED BY strmadmin DEFAULT TABLESPACE users QUOTA UNLIMITED ON users;
GRANT DBA, CONNECT, RESOURCE, SELECT_CATALOG_ROLE TO strmadmin;
GRANT EXECUTE ON DBMS_AQADM TO strmadmin;
GRANT EXECUTE ON DBMS_CAPTURE_ADM TO strmadmin;
GRANT EXECUTE ON DBMS_PROPAGATION_ADM TO strmadmin;
GRANT EXECUTE ON DBMS_STREAMS_ADM TO strmadmin;
GRANT EXECUTE ON DBMS_APPLY_ADM TO strmadmin;
GRANT EXECUTE ON DBMS_FLASHBACK TO strmadmin;
GRANT ALL ON scott.dept TO strmadmin;
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;
/
CONNECT strmadmin/strmadmin@DB1
EXEC DBMS_STREAMS_ADM.SET_UP_QUEUE(); -- 建立流队列
CREATE DATABASE LINK db2 CONNECT TO strmadmin IDENTIFIED BY strmadmin USING 'DB2'; -- 建立数据库连接
在目的库（DB2）重复上面2、3步骤。
4. LogMinor 表空间设置
CONN sys/password@DB1 AS SYSDBA
CREATE TABLESPACE logmnr_ts DATAFILE 'E:ORACLEPRODUCT10.1.0ORADATADB1logmnr01.dbf' SIZE 25 M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
EXECUTE DBMS_LOGMNR_D.SET_TABLESPACE('logmnr_ts');
5. SUPPLEMENTAL 日志
CONN sys/password@DB1 AS SYSDBA
ALTER TABLE scott.dept ADD SUPPLEMENTAL LOG GROUP log_group_dept_pk (deptno) ALWAYS;
6. 配置传播进程
CONNECT strmadmin/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;
/
-- 检查传播进程作业
SELECT job,TO_CHAR(last_date, 'DD-Mon-YYYY HH24:MI:SS') last_date,TO_CHAR(next_date, 'DD-Mon-YYYY HH24:MI:SS') next_date,what FROM dba_jobs;
7. 配置捕捉进程
CONNECT strmadmin/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 日志也被迁移。因为捕捉进程还没有启动，所以可以删除这些日志：
CONN sys/password@DB2 AS SYSDBA
ALTER TABLE scott.dept DROP SUPPLEMENTAL LOG GROUP log_group_dept_pk;
用 DBMS_APPLY_ADM 包设置 SCN
CONNECT strmadmin/strmadmin@db1
DECLARE
v_scn NUMBER;
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. 配置应用进程
CONNECT strmadmin/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. 启动应用进程
CONNECT strmadmin/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. 启动捕捉进程
CONNECT strmadmin/strmadmin@DB1
BEGIN
DBMS_CAPTURE_ADM.START_CAPTURE(
capture_name => 'capture_simp');
END;
/
12. 测试
-- 测试 DML
CONNECT scott/tiger@db1
INSERT INTO dept (deptno, dname, loc) VALUES (99, 'Test Dept', 'UK');
COMMIT;
SELECT * FROM dept;
DEPTNO DNAME LOC
---------- -------------- -------------
10 ACCOUNTING NEW YORK
20 RESEARCH DALLAS
30 SALES CHICAGO
40 OPERATIONS BOSTON
99 Test Dept UK
5 rows selected.
CONNECT scott/tiger@db2
SELECT * FROM dept;
DEPTNO DNAME LOC
---------- -------------- -------------
10 ACCOUNTING NEW YORK
20 RESEARCH DALLAS
30 SALES CHICAGO
40 OPERATIONS BOSTON
99 Test Dept UK
5 rows selected.
-- 测试 DDL
CONNECT scott/tiger@db1
ALTER TABLE dept ADD (new_col NUMBER(10))
/
DESC dept
Name Null? Type
---------------------------- -------- --------------
DEPTNO NOT NULL NUMBER(2)
DNAME VARCHAR2(14)
LOC VARCHAR2(13)
NEW_COL NUMBER(10)
CONNECT scott/tiger@db2
DESC dept
Name Null? Type
---------------------------- -------- --------------
DEPTNO NOT NULL NUMBER(2)
DNAME VARCHAR2(14)
LOC VARCHAR2(13)
NEW_COL NUMBER(10)
-- 可以用下列语句查看流的内容
CONNECT strmadmin/strmadmin@DB1
SELECT s.user_data.getTypeName() FROM streams_queue_table s;
SET SERVEROUTPUT ON
DECLARE
v_anydata SYS.ANYDATA;
v_lcr SYS.LCR$_ROW_RECORD;
v_row_list SYS.LCR$_ROW_LIST;
v_result PLS_INTEGER;
BEGIN
SELECT user_data
INTO v_anydata
FROM strmadmin.streams_queue_table
WHERE rownum < 2;
v_result := ANYDATA.GetObject(
self => v_anydata,
obj => v_lcr);
DBMS_OUTPUT.PUT_LINE('Command Type : ' || v_lcr.Get_Command_Type);
DBMS_OUTPUT.PUT_LINE('Object Owner : ' || v_lcr.Get_Object_Owner);
DBMS_OUTPUT.PUT_LINE('Object Name : ' || v_lcr.Get_Object_Name);
DBMS_OUTPUT.PUT_LINE('Source Database Name : ' || v_lcr.Get_Source_Database_Name);
END;
/
13. 清理
-- 可以使用下面的语句识别并删除所有规则
conn / as sysdba
BEGIN
FOR cur_rec IN (SELECT rule_owner, rule_name FROM dba_rules) LOOP
DBMS_RULE_ADM.DROP_RULE(
rule_name => cur_rec.rule_owner || '.' || cur_rec.rule_name,
force => TRUE);
END LOOP;
END;
/
-- 识别、停止并删除所有捕捉进程和应用进程
conn / as sysdba
BEGIN
FOR cur_rec IN (SELECT capture_name FROM dba_capture) LOOP
DBMS_CAPTURE_ADM.STOP_CAPTURE(
capture_name => cur_rec.capture_name);
DBMS_CAPTURE_ADM.DROP_CAPTURE(
capture_name => cur_rec.capture_name);
END LOOP;
FOR cur_rec IN (SELECT apply_name FROM dba_apply) LOOP
DBMS_APPLY_ADM.STOP_APPLY(
apply_name => cur_rec.apply_name);
DBMS_APPLY_ADM.DROP_APPLY(
apply_name => cur_rec.apply_name);
END LOOP;
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)
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. 实例设置
CONN sys/password@DB1 AS SYSDBA
create pfile from spfile;
-- 编辑 initsid.ora 文件，增加以下参数：
-- AQ_TM_PROCESSES=1;
-- GLOBAL_NAMES=TRUE;
SHUTDOWN IMMEDIATE;
create spfile from pfile;
STARTUP mount;
alter database archivelog; -- 改为归档模式
alter database open;
3. 流管理员设置
CONN sys/password@DB1 AS SYSDBA
CREATE USER strmadmin IDENTIFIED BY strmadmin DEFAULT TABLESPACE users QUOTA UNLIMITED ON users;
GRANT DBA, CONNECT, RESOURCE, SELECT_CATALOG_ROLE TO strmadmin;
GRANT EXECUTE ON DBMS_AQADM TO strmadmin;
GRANT EXECUTE ON DBMS_CAPTURE_ADM TO strmadmin;
GRANT EXECUTE ON DBMS_PROPAGATION_ADM TO strmadmin;
GRANT EXECUTE ON DBMS_STREAMS_ADM TO strmadmin;
GRANT EXECUTE ON DBMS_APPLY_ADM TO strmadmin;
GRANT EXECUTE ON DBMS_FLASHBACK TO strmadmin;
GRANT ALL ON scott.dept TO strmadmin;
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;
/
CONNECT strmadmin/strmadmin@DB1
EXEC DBMS_STREAMS_ADM.SET_UP_QUEUE(); -- 建立流队列
CREATE DATABASE LINK db2 CONNECT TO strmadmin IDENTIFIED BY strmadmin USING 'DB2'; -- 建立数据库连接
在目的库（DB2）重复上面2、3步骤。
4. LogMinor 表空间设置
CONN sys/password@DB1 AS SYSDBA
CREATE TABLESPACE logmnr_ts DATAFILE 'E:ORACLEPRODUCT10.1.0ORADATADB1logmnr01.dbf' SIZE 25 M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
EXECUTE DBMS_LOGMNR_D.SET_TABLESPACE('logmnr_ts');
5. SUPPLEMENTAL 日志
CONN sys/password@DB1 AS SYSDBA
ALTER TABLE scott.dept ADD SUPPLEMENTAL LOG GROUP log_group_dept_pk (deptno) ALWAYS;
6. 配置传播进程
CONNECT strmadmin/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;
/
-- 检查传播进程作业
SELECT job,TO_CHAR(last_date, 'DD-Mon-YYYY HH24:MI:SS') last_date,TO_CHAR(next_date, 'DD-Mon-YYYY HH24:MI:SS') next_date,what FROM dba_jobs;
7. 配置捕捉进程
CONNECT strmadmin/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 日志也被迁移。因为捕捉进程还没有启动，所以可以删除这些日志：
CONN sys/password@DB2 AS SYSDBA
ALTER TABLE scott.dept DROP SUPPLEMENTAL LOG GROUP log_group_dept_pk;
用 DBMS_APPLY_ADM 包设置 SCN
CONNECT strmadmin/strmadmin@db1
DECLARE
v_scn NUMBER;
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. 配置应用进程
CONNECT strmadmin/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. 启动应用进程
CONNECT strmadmin/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. 启动捕捉进程
CONNECT strmadmin/strmadmin@DB1
BEGIN
DBMS_CAPTURE_ADM.START_CAPTURE(
capture_name => 'capture_simp');
END;
/
12. 测试
-- 测试 DML
CONNECT scott/tiger@db1
INSERT INTO dept (deptno, dname, loc) VALUES (99, 'Test Dept', 'UK');
COMMIT;
SELECT * FROM dept;
DEPTNO DNAME LOC
---------- -------------- -------------
10 ACCOUNTING NEW YORK
20 RESEARCH DALLAS
30 SALES CHICAGO
40 OPERATIONS BOSTON
99 Test Dept UK
5 rows selected.
CONNECT scott/tiger@db2
SELECT * FROM dept;
DEPTNO DNAME LOC
---------- -------------- -------------
10 ACCOUNTING NEW YORK
20 RESEARCH DALLAS
30 SALES CHICAGO
40 OPERATIONS BOSTON
99 Test Dept UK
5 rows selected.
-- 测试 DDL
CONNECT scott/tiger@db1
ALTER TABLE dept ADD (new_col NUMBER(10))
/
DESC dept
Name Null? Type
---------------------------- -------- --------------
DEPTNO NOT NULL NUMBER(2)
DNAME VARCHAR2(14)
LOC VARCHAR2(13)
NEW_COL NUMBER(10)
CONNECT scott/tiger@db2
DESC dept
Name Null? Type
---------------------------- -------- --------------
DEPTNO NOT NULL NUMBER(2)
DNAME VARCHAR2(14)
LOC VARCHAR2(13)
NEW_COL NUMBER(10)
-- 可以用下列语句查看流的内容
CONNECT strmadmin/strmadmin@DB1
SELECT s.user_data.getTypeName() FROM streams_queue_table s;
SET SERVEROUTPUT ON
DECLARE
v_anydata SYS.ANYDATA;
v_lcr SYS.LCR$_ROW_RECORD;
v_row_list SYS.LCR$_ROW_LIST;
v_result PLS_INTEGER;
BEGIN
SELECT user_data
INTO v_anydata
FROM strmadmin.streams_queue_table
WHERE rownum < 2;
v_result := ANYDATA.GetObject(
self => v_anydata,
obj => v_lcr);
DBMS_OUTPUT.PUT_LINE('Command Type : ' || v_lcr.Get_Command_Type);
DBMS_OUTPUT.PUT_LINE('Object Owner : ' || v_lcr.Get_Object_Owner);
DBMS_OUTPUT.PUT_LINE('Object Name : ' || v_lcr.Get_Object_Name);
DBMS_OUTPUT.PUT_LINE('Source Database Name : ' || v_lcr.Get_Source_Database_Name);
END;
/
13. 清理
-- 可以使用下面的语句识别并删除所有规则
conn / as sysdba
BEGIN
FOR cur_rec IN (SELECT rule_owner, rule_name FROM dba_rules) LOOP
DBMS_RULE_ADM.DROP_RULE(
rule_name => cur_rec.rule_owner || '.' || cur_rec.rule_name,
force => TRUE);
END LOOP;
END;
/
-- 识别、停止并删除所有捕捉进程和应用进程
conn / as sysdba
BEGIN
FOR cur_rec IN (SELECT capture_name FROM dba_capture) LOOP
DBMS_CAPTURE_ADM.STOP_CAPTURE(
capture_name => cur_rec.capture_name);
DBMS_CAPTURE_ADM.DROP_CAPTURE(
capture_name => cur_rec.capture_name);
END LOOP;
FOR cur_rec IN (SELECT apply_name FROM dba_apply) LOOP
DBMS_APPLY_ADM.STOP_APPLY(
apply_name => cur_rec.apply_name);
DBMS_APPLY_ADM.DROP_APPLY(
apply_name => cur_rec.apply_name);
END LOOP;
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)
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- Oracle 多主（Master Replication）复制配置
- Oracle 高级数据复制（Advanced Replication）
- Oracle自定义一个把 IP 地址转成整数的函数
- Oracle 自定义split 函数
- 线上VMS业务oracle数据库用户XXX密码更改
- RAC一个节点自动重启问题分析
- Oracle中的三种 UPDATE FROM 的解决方案
- 一个用 Oracle 函数索引进行优化的例子
- wm_concat函数 用法
- Oracle之表间关联查询
- Oracle 中 isnumeric 的三种实现
- 配置 Oracle 透明网关访问 SQL Server
- Oracle 列数据聚合方法汇总
- Oracle小函数大用处！
- TEC1303.Form个性化技术总结 - 第四部分 Custom.pll个性化开发实例
- Oracle 索引的分类
- oracle 递归查询学习
- [转载Oracle官方中文博客]关于RunQ过高引起的latch等待问题
- Oracle 11g 闪回作业
- Oracle sqlldr 如何导入一个日期列