9206下配置双向stream
一个客户要在9206上用双向stream，个人不推荐这么做（每天800G的日志有点恐怖），今天下午给搭建了测试环境，配置过程记录下。
SQL> select * from v$version;
BANNER
----------------------------------------------------------------
Oracle9i Enterprise Edition Release 9.2.0.6.0 - 64bit Production
PL/SQL Release 9.2.0.6.0 - Production
CORE 9.2.0.6.0 Production
TNS for IBM/AIX RISC System/6000: Version 9.2.0.6.0 - Production
NLSRTL Version 9.2.0.6.0 - Production
5 rows selected.
/***************************************************
--需要在2个数据库分别执行：
conn /as sysdba
show parameter JOB_QUEUE_PROCESSES
show parameter AQ_TM_PROCESSES
show parameter LOG_PARALLELISM
show parameter global_name
create tablespace pmid datafile '/oracle/app/oracle/oradata/crmtest/pmid.dbf' size 512m;
create user strmadmin identified by strmadmin default tablespace pmid;
grant dba, CONNECT, RESOURCE, SELECT_CATALOG_ROLE to strmadmin;
GRANT EXECUTE ON DBMS_APPLY_ADM 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;
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;
/
create user strmtest identified by strmtest default tablespace pmid;
grant dba to strmtest;
--create tablespace PMID(修改名字) for logmnr --创建logmnr表空间
EXECUTE DBMS_LOGMNR_D.SET_TABLESPACE('PMID');
alter system set global_names=true scope=both;
/***************************************************
--只在source数据库执行；
SQL> conn strmtest/strmtest
create table test1(id number not null primary key,mc varchar2(200));
create table test2(id number,mc varchar2(200));
/***************************************************
在source数据库上执行
conn /as sysdba
ALTER TABLE strmtest.test1 ADD (time TIMESTAMP WITH TIME ZONE);
ALTER TABLE strmtest.test2 ADD (time TIMESTAMP WITH TIME ZONE);
CREATE OR REPLACE TRIGGER strmtest.insert_time_test1
BEFORE
INSERT OR UPDATE ON strmtest.test1 FOR EACH ROW
BEGIN
IF :OLD.TIME IS NULL OR :OLD.TIME < SYSTIMESTAMP THEN
:NEW.TIME := SYSTIMESTAMP;
ELSE
:NEW.TIME := :OLD.TIME + 1 / 86400;
END IF;
END;
/
CREATE OR REPLACE TRIGGER strmtest.insert_time_test2
BEFORE
INSERT OR UPDATE ON strmtest.test2 FOR EACH ROW
BEGIN
IF :OLD.TIME IS NULL OR :OLD.TIME < SYSTIMESTAMP THEN
:NEW.TIME := SYSTIMESTAMP;
ELSE
:NEW.TIME := :OLD.TIME + 1 / 86400;
END IF;
END;
/
ALTER TABLE strmtest.test1 ADD SUPPLEMENTAL LOG GROUP log_group_test1(id, mc, time) ALWAYS;
ALTER TABLE strmtest.test2 ADD SUPPLEMENTAL LOG GROUP log_group_test2(id, mc, time) ALWAYS;
GRANT ALL ON strmtest.test1 TO strmadmin;
GRANT ALL ON strmtest.test2 TO strmadmin;
/***************************************************
--在source数据库执行：
conn strmadmin/strmadmin
CREATE DATABASE LINK test CONNECT TO strmadmin IDENTIFIED BY strmadmin USING 'test';
EXEC DBMS_STREAMS_ADM.SET_UP_QUEUE(queue_name=>'capture_queue');
EXEC DBMS_STREAMS_ADM.SET_UP_QUEUE(queue_name=>'apply_queue');
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(
table_name => 'strmtest.test1',
streams_name => 'stream_propagation',
source_queue_name => 'strmadmin.capture_queue',
destination_queue_name => 'strmadmin.apply_queue@test',
include_dml => true,
include_ddl => true,
source_database => 'crmtest');
END;
/
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(
table_name => 'strmtest.test2',
streams_name => 'stream_propagation',
source_queue_name => 'strmadmin.capture_queue',
destination_queue_name => 'strmadmin.apply_queue@test',
include_dml => true,
include_ddl => true,
source_database => 'crmtest');
END;
/
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name => 'strmtest.test1',
streams_type => 'capture',
streams_name => 'stream_capture',
queue_name => 'strmadmin.capture_queue',
include_dml => true,
include_ddl => true);
END;
/
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name => 'strmtest.test2',
streams_type => 'capture',
streams_name => 'stream_capture',
queue_name => 'strmadmin.capture_queue',
include_dml => true,
include_ddl => true);
END;
/
DECLARE
cols DBMS_UTILITY.NAME_ARRAY;
BEGIN
cols(1) := 'id';
cols(2) := 'mc';
cols(3) := 'time';
DBMS_APPLY_ADM.SET_UPDATE_CONFLICT_HANDLER(
object_name => 'strmtest.test1',
method_name => 'MAXIMUM',
resolution_column => 'time',
column_list => cols);
END;
/
DECLARE
cols DBMS_UTILITY.NAME_ARRAY;
BEGIN
cols(1) := 'id';
cols(2) := 'mc';
cols(3) := 'time';
DBMS_APPLY_ADM.SET_UPDATE_CONFLICT_HANDLER(
object_name => 'strmtest.test2',
method_name => 'MAXIMUM',
resolution_column => 'time',
column_list => cols);
END;
/
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name => 'strmtest.test1',
streams_type => 'apply',
streams_name => 'stream_apply',
queue_name => 'strmadmin.apply_queue',
include_dml => true,
include_ddl => true,
source_database => 'test');
END;
/
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name => 'strmtest.test2',
streams_type => 'apply',
streams_name => 'stream_apply',
queue_name => 'strmadmin.apply_queue',
include_dml => true,
include_ddl => true,
source_database => 'test');
END;
/
exp userid=strmtest/strmtest@crmtest FILE=streams.dmp TABLES=strmtest.test1,strmtest.test2 OBJECT_CONSISTENT=y
imp userid=strmtest/strmtest@test full=y FILE=streams.dmp COMMIT=y LOG=import.log STREAMS_INSTANTIATION=y
/***************************************************
--在target数据库执行：
CONN /AS SYSDBA
GRANT ALL ON strmtest.test1 TO strmadmin;
GRANT ALL ON strmtest.test2 TO strmadmin;
--确认SUPPLEMENTAL LOG和trigge和权限都正常
CONNECT strmadmin/strmadmin@test
CREATE DATABASE LINK crmtest CONNECT TO strmadmin IDENTIFIED BY strmadmin USING 'crmtest';
EXEC DBMS_STREAMS_ADM.SET_UP_QUEUE(queue_name=>'capture_queue');
EXEC DBMS_STREAMS_ADM.SET_UP_QUEUE(queue_name=>'apply_queue');
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(
table_name => 'strmtest.test1',
streams_name => 'stream_propagation',
source_queue_name => 'strmadmin.capture_queue',
destination_queue_name => 'strmadmin.apply_queue@crmtest',
include_dml => true,
include_ddl => true,
source_database => 'test');
END;
/
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_PROPAGATION_RULES(
table_name => 'strmtest.test2',
streams_name => 'stream_propagation',
source_queue_name => 'strmadmin.capture_queue',
destination_queue_name => 'strmadmin.apply_queue@crmtest',
include_dml => true,
include_ddl => true,
source_database => 'test');
END;
/
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name => 'strmtest.test1',
streams_type => 'capture',
streams_name => 'stream_capture',
queue_name => 'strmadmin.capture_queue',
include_dml => true,
include_ddl => true);
END;
/
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name => 'strmtest.test2',
streams_type => 'capture',
streams_name => 'stream_capture',
queue_name => 'strmadmin.capture_queue',
include_dml => true,
include_ddl => true);
END;
/
DECLARE
cols DBMS_UTILITY.NAME_ARRAY;
BEGIN
cols(1) := 'id';
cols(2) := 'mc';
cols(3) := 'time';
DBMS_APPLY_ADM.SET_UPDATE_CONFLICT_HANDLER(
object_name => 'strmtest.test1',
method_name => 'MAXIMUM',
resolution_column => 'time',
column_list => cols);
END;
/
DECLARE
cols DBMS_UTILITY.NAME_ARRAY;
BEGIN
cols(1) := 'id';
cols(2) := 'mc';
cols(3) := 'time';
DBMS_APPLY_ADM.SET_UPDATE_CONFLICT_HANDLER(
object_name => 'strmtest.test2',
method_name => 'MAXIMUM',
resolution_column => 'time',
column_list => cols);
END;
/
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name => 'strmtest.test1',
streams_type => 'apply',
streams_name => 'stream_apply',
queue_name => 'strmadmin.apply_queue',
include_dml => true,
include_ddl => true,
source_database => 'crmtest');
END;
/
BEGIN
DBMS_STREAMS_ADM.ADD_TABLE_RULES(
table_name => 'strmtest.test2',
streams_type => 'apply',
streams_name => 'stream_apply',
queue_name => 'strmadmin.apply_queue',
include_dml => true,
include_ddl => true,
source_database => 'crmtest');
END;
/
DECLARE
v_scn NUMBER;
BEGIN
v_scn := DBMS_FLASHBACK.GET_SYSTEM_CHANGE_NUMBER();
DBMS_APPLY_ADM.SET_TABLE_INSTANTIATION_SCN@crmtest(
source_object_name => 'strmtest.test1',
source_database_name => 'test',
instantiation_scn => v_scn);
DBMS_APPLY_ADM.SET_TABLE_INSTANTIATION_SCN@crmtest(
source_object_name => 'strmtest.test2',
source_database_name => 'test',
instantiation_scn => v_scn);
END;
/
exec dbms_apply_adm.start_apply('stream_apply');
/***************************************************
--在source数据库执行：
exec dbms_capture_adm.start_capture('stream_capture');
exec dbms_apply_adm.start_apply('stream_apply');
/***************************************************
--在target数据库执行：
exec dbms_capture_adm.start_capture('stream_capture');
测试：
在source数据库insert数据：
SQL> insert into strmtest.test1(id,mc) values(1,'111111');
1 row created.
SQL> commit;
Commit complete.
在target数据库查询：
SQL> select * from strmtest.test1;
ID MC TIME
---------- ---------- --------------------------------------------------
1 111111 29-MAY-08 10.07.05.549004 PM +08:00
在target数据库insert数据：
SQL> insert into strmtest.test1(id,mc) values(2,'222222');
1 row created.
SQL> commit;
Commit complete.
在source数据库查询：
SQL> select * from strmtest.test1;
ID MC TIME
---------- ---------- --------------------------------------------------
1 111111 29-MAY-08 10.07.05.549004 PM +08:00
2 222222 29-MAY-08 10.08.10.132493 PM +08:00
在source库insert数据：
SQL> insert into strmtest.test1(id,mc) values(3,'333333');
1 row created.
SQL> commit;
Commit complete.
在target库查询数据：
SQL> select * from strmtest.test1;
ID MC TIME
---------- ---------- --------------------------------------------------
1 111111 29-MAY-08 10.07.05.549004 PM +08:00
2 222222 29-MAY-08 10.08.10.132493 PM +08:00
3 333333 29-MAY-08 10.14.22.920806 PM +08:00
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/756652/viewspace-324236/ ，如需转载，请注明出处，否则将追究法律责任。
