Oracle 11g streams部署

環境

	源服務器	目標服務器
系統版本	CentOS Linux release 7.3.1611 (Core)	CentOS Linux release 7.3.1611 (Core)
主機名	sht-sgmhadoopdn-02	sht-sgmhadoopdn-03
數據庫版本	EE 11.2.0.4.0	EE 11.2.0.4.0
dbname	FINMART	FINMART
global_name	FINMART1	FINMART2

Streams流復制技術介紹

Streams通過logmnr（日志挖掘）技術從oracle的log中解析出數據，然后傳遞到目標庫並應用，從而將源庫的數據復制到目標庫。在Stream 環境下，復制的起點數據庫叫作Source Database， 復制的終點數據庫叫作Target Database。 在這兩個數據庫上都要創建一個隊列，其中的Source Database上的是發送隊列，而Target Database上的是接收隊列。

數據庫的所有操作都會被記錄在日志中。 配好Stream環境后， 在Source Database上會有一個捕獲進程(Capture Process), 該進程利用Logminer技術從日志中提取DDL,DML語句，這些語句用一種特殊的格式表達，叫作邏輯變更記錄(Logical Change Record, LCR). 一個LCR對應一個原子的行變更，因此源數據庫上的一個DML語句，可能對應若干個LCR記錄。 這些LCR會保存到Sourece Database的本地發送隊列中。然后傳播進程(Propagation Process)把這些記錄通過網絡發送到Target Database的接收隊列。 在Target Database上會有一個應用進程(Apply Process)， 這個進程從本地的接收隊列中取出LCR記錄，然后在本地應用，實現數據同步 。

整個的復制過程可以分成三個步驟：捕獲(capture)，傳播(propagation)和應用(apply)，利用高級隊列(advance queue)來將這三個步驟的數據串起來，通過在步驟中定義不同的規則(rule)來控制需要復制的數據。復制可以基於全庫，基於表空間，基於用戶或者基於表，提供了相當大的靈活性。

如下圖所示，通過oracle streams技術在兩個數據庫之間復制一張數據表，在源服務器str1.example.com上capture進程在hr的schema中捕捉 jobs表的DDL和DML語句並記錄到發送隊列（QUEUE）中，propagation進程將這些變更傳播到目標服務器 str2.example.com的接收隊列中，apply進程負責將這些變更應用到目標數據庫中。

 

本次計划實現目標：

假定目標數據庫目前是正常運行狀態，將源庫中have用戶的schema實時同步到目標庫上，即用戶級別的復制。

1.環境准備

1.1 源庫和目標庫均修改為歸檔模式

源庫

SQL> archive log list;
Database log mode           Archive Mode
Automatic archival           Enabled
Archive destination           /arch/oradata/FINMART
Oldest online log sequence     80
Next log sequence to archive   83
Current log sequence           83

目標庫

SQL> archive log list;
Database log mode           Archive Mode
Automatic archival           Enabled
Archive destination           /arch/oradata/FINMART
Oldest online log sequence     1
Next log sequence to archive   1
Current log sequence           1

1.2 修改源庫和目標庫的參數

源庫

SQL> alter system set aq_tm_processes=2 scope=both;
SQL> alter system set global_names=true scope=both;
SQL> alter system set job_queue_processes=10 scope=both; 
SQL> alter system set streams_pool_size=200M scope=both;
SQL> alter system set open_links=4 scope=spfile;
SQL> alter database rename GLOBAL_NAME to "FINMART1";

• • job_queue_processes 　　決定了job作業能夠使用的總進程數
  • aq_tm_processes               該參數決定了數據庫啟動時Qnnn進程的數量，負責監視高級隊列和負責隊列傳播（propagation）
  • streams_pool_size　　      手動指定srteams池的大小
  • open_links                         每個session最多允許的dblink數量
  • global_names                    streams必須要設置該參數為true，當GLOBAL_NAMES參數設置為TRUE時，創建DBLINK的名稱必須與被連接庫的GLOBAL_NAME一致。

 注意：如果源庫和目標庫的dbname相同，說明源庫和目標庫的global_name也相同，這時需要將源庫和目標庫的global_name修改為不一致，否則創建db link會報錯

目標庫

SQL> alter system set aq_tm_processes=2 scope=both;
SQL> alter system set global_names=true scope=both;
SQL> alter system set job_queue_processes=10 scope=both; 
SQL> alter system set streams_pool_size=200M scope=both;
SQL> alter system set open_links=4 scope=spfile;
SQL> alter database rename GLOBAL_NAME to "FINMART2";

 1.3 在源庫和目標庫創建srteams管理用戶和測試用戶

SQL> create tablespace streams_tbs datafile '/db01/oradata/FINMART/streams_tbs01.dbf' size 100m reuse autoextend on maxsize unlimited;
SQL> create user strmadmin identified by strmadmin default tablespace streams_tbs quota unlimited on streams_tbs;
SQL> grant dba to strmadmin;
SQL> exec DBMS_STREAMS_AUTH.GRANT_ADMIN_PRIVILEGE('strmadmin');

SQL> create tablespace dave_tbs datafile '/db01/oradata/FINMART/dave_tbs01.dbf' size 100M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;
SQL> create user dave identified by dave default tablespace dave_tbs quota unlimited on dave_tbs;

 

SQL> create table dave.tb1(id int primary key,name varchar2(40),value varchar2(40));
SQL> insert into dave.tb1(id,name,value) values('1','birthday','19');
SQL> commit;

1.4 在源庫和目標庫創建tnsnames

FINMART1 =
  (DESCRIPTION =
    (ADDRESS = (PROTOCOL = TCP)(HOST = sht-sgmhadoopdn-02)(PORT = 1521))
    (CONNECT_DATA =
      (SERVER = DEDICATED)
      (SERVICE_NAME = FINMART)
    )
  )

FINMART2 =
  (DESCRIPTION =
    (ADDRESS = (PROTOCOL = TCP)(HOST = sht-sgmhadoopdn-03)(PORT = 1521))
    (CONNECT_DATA =
      (SERVER = DEDICATED)
      (SERVICE_NAME = FINMART)
    )
  )

1.5 在源庫和目標庫創建db link

源庫

SQL> connect strmadmin/strmadmin@finmart1
SQL> create database link finmart2 connect to strmadmin identified by strmadmin using 'finmart2';
SQL> select * from global_name@finmart2;

GLOBAL_NAME
--------------------------------------------------------------------------------
FINMART2

 目標庫

SQL> connect strmadmin/strmadmin@finmart2
SQL> create database link finmart1 connect to strmadmin identified by strmadmin using 'finmart1';
SQL> select * from global_name@finmart1;

GLOBAL_NAME
--------------------------------------------------------------------------------
FINMART1

注意：database link名稱必須是目標庫的global_name名稱

1.6 在數據庫上啟動追加日志

附加日志（supplemental log）可以指示數據庫在日志中添加額外信息到日志流中，以支持基於日志的工具，如邏輯standby、streams、GoldenGate、LogMiner。可以在數據庫和表上設置。

SQL> alter database add supplemental log data;

 2. 在源庫上創建發送隊列

SQL> BEGIN
  2  DBMS_STREAMS_ADM.SET_UP_QUEUE(
  3  queue_table => 'SOURCE_QUEUE_TABLE',
  4  queue_name => 'SOURCE_QUEUE',
  5  queue_user => 'strmadmin');
  6  END;
  7  /

PL/SQL procedure successfully completed.

• queue_name 隊列名稱
• queue_table 隊列表名
• queue_user 隊列所有者

通過數據字典dba_queues、dba_queue_tables查看 

column NAME format a40
column QUEUE_TABLE format a20
column OWNER format a20
select NAME,QUEUE_TABLE,OWNER,QUEUE_TYPE from dba_queues where OWNER='STRMADMIN';

NAME                     　　　　　　　　　 QUEUE_TABLE          OWNER                QUEUE_TYPE
---------------------------------------- -------------------- -------------------- ------------------------------------------------------------
AQ$_SOURCE_QUEUE_TABLE_E         　　　 　SOURCE_QUEUE_TABLE   STRMADMIN            EXCEPTION_QUEUE
SOURCE_QUEUE                 　　　　　　　SOURCE_QUEUE_TABLE   STRMADMIN            NORMAL_QUEUE

column OBJECT_TYPE format a20
select OWNER,QUEUE_TABLE,OBJECT_TYPE from dba_queue_tables where owner='STRMADMIN';

OWNER                QUEUE_TABLE          OBJECT_TYPE
-------------------- -------------------- --------------------
STRMADMIN            SOURCE_QUEUE_TABLE   SYS.ANYDATA

如果刪除發送隊列，命令如下

exec dbms_streams_adm.remove_queue(queue_name => 'source_queue',cascade => true,drop_unused_queue_table => true);

select NAME,QUEUE_TABLE,OWNER,QUEUE_TYPE from dba_queues where OWNER='STRMADMIN';

no rows selected
select OWNER,QUEUE_TABLE,OBJECT_TYPE from dba_queue_tables where owner='STRMADMIN';

no rows selected

3. 在目標庫上創建接收隊列

SQL> BEGIN
  2  DBMS_STREAMS_ADM.SET_UP_QUEUE(
  3  queue_table => 'TARGET_QUEUE_TABLE',
  4  queue_name => 'TARGET_QUEUE',
  5  queue_user => 'strmadmin');
  6  END;
  7  /

PL/SQL procedure successfully completed.

select NAME,QUEUE_TABLE,OWNER,QUEUE_TYPE from dba_queues where OWNER='STRMADMIN';

NAME                     　　　　　　　　　 QUEUE_TABLE          OWNER                QUEUE_TYPE
---------------------------------------- -------------------- -------------------- ------------------------------------------------------------
TARGET_QUEUE                 　　　　　　　TARGET_QUEUE_TABLE   STRMADMIN            NORMAL_QUEUE
AQ$_TARGET_QUEUE_TABLE_E                 TARGET_QUEUE_TABLE   STRMADMIN            EXCEPTION_QUEUE

select OWNER,QUEUE_TABLE,OBJECT_TYPE from dba_queue_tables where owner='STRMADMIN';

OWNER                QUEUE_TABLE          OBJECT_TYPE
-------------------- -------------------- --------------------
STRMADMIN            TARGET_QUEUE_TABLE   SYS.ANYDATA

4. 在源庫上創建capture 進程

SQL> BEGIN
  2  DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
  3  schema_name => 'dave',
  4  streams_type => 'capture',
  5  streams_name => 'capture_stream',
  6  queue_name => 'strmadmin.SOURCE_QUEUE',
  7  include_dml => true,
  8  include_ddl => true,
  9  source_database => 'FINMART1',
 10  include_tagged_lcr => false,
 11  inclusion_rule => true);
 12  END;
 13  /

PL/SQL procedure successfully completed.

column CAPTURE_NAME format a20
column QUEUE_NAME format a20
column QUEUE_OWNER format a20column CAPTURE_USER format a20
column SOURCE_DATABASE format a20
column CAPTURE_TYPE format a20
column START_TIME format a40
select CAPTURE_NAME,QUEUE_NAME,QUEUE_OWNER,CAPTURE_USER,START_SCN,STATUS,SOURCE_DATABASE,CAPTURE_TYPE,START_TIME from dba_capture;

CAPTURE_NAME         QUEUE_NAME           QUEUE_OWNER          CAPTURE_USER         START_SCN  STATUS                   SOURCE_DATABASE      CAPTURE_TYPE                   START_TIME
-------------------- -------------------- -------------------- -------------------- ---------- ------------------------ -------------------- ------------------------------ ---------------------------------------------------------------------------
CAPTURE_STREAM       SOURCE_QUEUE         STRMADMIN            STRMADMIN            7082908    DISABLED                 FINMART1             LOCAL                          16-SEP-18 06.10.18.000000 PM

column SCHEMA_NAME format a20
select * from dba_capture_prepared_schemas;

SCHEMA_NAME              TIMESTAMP       SUPPLEMENTAL_LOG_DATA_PK SUPPLEMENTAL_LOG_DATA_UI SUPPLEMENTAL_LOG_DATA_FK SUPPLEMENTAL_LOG_DATA_AL
------------------------ --------------- ------------------------ ------------------------ ------------------------ ------------------------
DAVE                     16-SEP-18       IMPLICIT                 IMPLICIT                 IMPLICIT                 NO

5.在源庫上創建傳播進程（Propagation Process）

SQL> BEGIN
  2  DBMS_STREAMS_ADM.ADD_SCHEMA_PROPAGATION_RULES(
  3  schema_name => 'dave',
  4  streams_name => 'source_to_target',
  5  source_queue_name => 'strmadmin.SOURCE_QUEUE',
  6  destination_queue_name => 'strmadmin.TARGET_QUEUE@FINMART2',
  7  include_dml => true,
  8  include_ddl => true,
  9  source_database => 'FINMART1',
 10  inclusion_rule => true,
 11  queue_to_queue => true);
 12  END;
 13  /

PL/SQL procedure successfully completed.

column PROPAGATION_NAME format a30
column SOURCE_QUEUE_OWNER format a20
column SOURCE_QUEUE_NAME format a30
column DESTINATION_QUEUE_OWNER format a30
column DESTINATION_QUEUE_NAME format a30
column DESTINATION_DBLINK format a20
column RULE_SET_OWNER format a20
select PROPAGATION_NAME,SOURCE_QUEUE_OWNER,SOURCE_QUEUE_NAME,DESTINATION_QUEUE_OWNER,DESTINATION_QUEUE_NAME,DESTINATION_DBLINK,RULE_SET_OWNER,QUEUE_TO_QUEUE,STATUS from dba_propagation;

PROPAGATION_NAME               SOURCE_QUEUE_OWNER             SOURCE_QUEUE_NAME              DESTINATION_QUEUE_OWNER        DESTINATION_QUEUE_NAME         DESTINATION_DBLINK   RULE_SET_OWNER       QUEUE_TO_QUEUE  STATUS
------------------------------ ------------------------------ ------------------------------ ------------------------------ ------------------------------ -------------------- -------------------- --------------- ------------------------
SOURCE_TO_TARGET               STRMADMIN                      SOURCE_QUEUE                   STRMADMIN                      TARGET_QUEUE                   FINMART2             STRMADMIN            TRUE            ENABLED

6.修改propagation休眠時間為0，表示實時傳播LCR

SQL> BEGIN
  2  dbms_aqadm.alter_propagation_schedule(
  3  queue_name => 'SOURCE_QUEUE',
  4  destination => 'FINMART2',
  5  destination_queue => 'TARGET_QUEUE',
  6  latency => 0);
  7  END;
  8  /

PL/SQL procedure successfully completed.

7.在目標庫創建Apply進程

SQL> BEGIN
  2  DBMS_STREAMS_ADM.ADD_SCHEMA_RULES(
  3  schema_name => 'dave',
  4  streams_type => 'apply',
  5  streams_name => 'target_apply_stream',
  6  queue_name => 'strmadmin.TARGET_QUEUE',
  7  include_dml => true,
  8  include_ddl => true,
  9  include_tagged_lcr => false,
 10  source_database => 'FINMART1',
 11  inclusion_rule => true);
 12  END;
 13  /

PL/SQL procedure successfully completed.

column APPLY_NAME format a20
column QUEUE_NAME format a20
column QUEUE_OWNER format a20
column RULE_SET_OWNER format a20
column APPLY_USER format a20
column APPLY_DATABASE_LINK format a20
select APPLY_NAME,QUEUE_NAME,QUEUE_OWNER,APPLY_CAPTURED,RULE_SET_OWNER,APPLY_USER,APPLY_DATABASE_LINK,STATUS from dba_apply;

APPLY_NAME           QUEUE_NAME           QUEUE_OWNER          APPLY_CAP RULE_SET_OWNER       APPLY_USER           APPLY_DATABASE_LINK  STATUS
-------------------- -------------------- -------------------- --------- -------------------- -------------------- -------------------- ------------------------
TARGET_APPLY_STREAM  TARGET_QUEUE         STRMADMIN            YES       STRMADMIN            STRMADMIN                                 DISABLED

 8.實例化數目標數據庫

源庫

exp USERID=dave/dave@finmart1 file=dave.dmp object_consistent=y rows=y

目標庫

imp strmadmin/strmadmin@finmart2 file='dave.dmp' ignore=y commit=y log='dave.log' streams_instantiation=y fromuser=dave touser=dave

9. 在目標庫上啟動apply進程

SQL> BEGIN
  2  DBMS_APPLY_ADM.START_APPLY(
  3  apply_name => 'target_apply_stream');
  4  END;
  5  /

PL/SQL procedure successfully completed.

select APPLY_NAME,QUEUE_NAME,QUEUE_OWNER,APPLY_CAPTURED,RULE_SET_OWNER,APPLY_USER,APPLY_DATABASE_LINK,STATUS from dba_apply;

APPLY_NAME         QUEUE_NAME       QUEUE_OWNER           APPLY_CAP RULE_SET_OWNER       APPLY_USER       APPLY_DATABASE_LINK    STATUS
-------------------- -------------------- -------------------- --------- -------------------- -------------------- -------------------- ------------------------
TARGET_APPLY_STREAM  TARGET_QUEUE      STRMADMIN           YES     STRMADMIN          STRMADMIN                 ENABLED

10.在源庫上啟動capture進程

SQL> BEGIN
  2  DBMS_CAPTURE_ADM.START_CAPTURE(
  3  capture_name => 'capture_stream');
  4  END;
  5  /

PL/SQL procedure successfully completed.

select CAPTURE_NAME,QUEUE_NAME,QUEUE_OWNER,CAPTURE_USER,START_SCN,STATUS,SOURCE_DATABASE,CAPTURE_TYPE,START_TIME from dba_capture;

CAPTURE_NAME         QUEUE_NAME       QUEUE_OWNER           CAPTURE_USER         START_SCN STATUS            SOURCE_DATABASE      CAPTURE_TYPE            START_TIME
-------------------- -------------------- -------------------- -------------------- ---------- ------------------------ -------------------- ------------------------------ ---------------------------------------------------------------------------
CAPTURE_STREAM         SOURCE_QUEUE      STRMADMIN           STRMADMIN           7082908 ENABLED            FINMART1         LOCAL                16-SEP-18 06.10.18.000000 PM

11.驗證DML和DDL

select * from dave.tb1@finmart2;

    ID NAME                                         VALUE
---------- ---------------------------------------- ------------------------------------------------------------------------------------------------------------------------
     1 birthday                                     19insert into dave.tb1(id,name,value) values('2','sex','female');

1 row created.

commit;

Commit complete.

select * from dave.tb1@finmart2;

    ID NAME                                         VALUE
---------- ---------------------------------------- ------------------------------------------------------------------------------------------------------------------------
     1 birthday                                     19
     2 sex                                          female

SQL> create table dave.tb2 as select * from dba_objects;

Table created.

SQL> select count(*) from dave.tb2;

  COUNT(*)
----------
     86453

SQL> select count(*) from dave.tb2@finmart2;
select count(*) from dave.tb2@finmart2
                          *
ERROR at line 1:
ORA-00942: table or view does not exist
ORA-02063: preceding line from FINMART2


SQL> select count(*) from dave.tb2@finmart2;

  COUNT(*)
----------
     0

SQL> select count(*) from dave.tb2@finmart2;

  COUNT(*)
----------
     86453