Oracle CDC配置案例

异步部署

1. 环境的配置准备

1.1.    数据库版本

SQL> select * from v$version;
BANNER
-----------------------------------------------------------------
Oracle Database 11g Enterprise Edition Release 11.2.0.1. - Production
PL/SQL Release 11.2.0.1. - Production
CORE    11.2.0.1.      Production
TNS for Linux: Version 11.2.0.1. - Production
NLSRTL Version 11.2.0.1. - Production

1.2.    配置数据库参数

SQL> show parameter job_que
NAME                                 TYPE        VALUE
----------------------------------------------- job_queue_processes                  integer     

SQL> show parameter streams_pool_size
NAME                                 TYPE        VALUE
-----------------------------------------------
streams_pool_size                    big integer 

SQL> show parameter sga_targ
NAME                                 TYPE        VALUE
-----------------------------------------------
sga_target                           big integer 

SQL> show parameter memory_targ
NAME                                 TYPE        VALUE
----------------------------------------------
memory_target                        big integer 472M

SQL> show parameter java_pool
NAME                                 TYPE        VALUE
------------------------------------ -----------
java_pool_size                       big integer 

SQL> alter system set streams_pool_size=50m ;

System altered.

SQL> show parameter java_pool

NAME                                 TYPE        VALUE
------------------------------------ -----------
java_pool_size                       big integer
SQL> alter system set java_pool_size=50m;

System altered.

SQL> show parameter undo_ret

NAME                                 TYPE        VALUE
------------------------------------ -----------
undo_retention                       integer
SQL> alter system set undo_retention=;

System altered.

SQL> show parameter streams_pool
NAME                                 TYPE        VALUE
------------------------------------ -----------
streams_pool_size                    big integer 52M

SQL> show parameter java_pool
NAME                                 TYPE        VALUE
------------------------------------ -----------
java_pool_size                       big integer 52M

SQL> show parameter undo_re
NAME                                 TYPE        VALUE
------------------------------------ -----------
undo_retention                       integer     

1.3.    开启归档及补充日志

SQL> archive log list
Database log mode              Archive Mode
Automatic archival             Enabled
Archive destination            /u02/archivelog_dest
Oldest online log sequence
Next log sequence to archive
Current log sequence           

SQL> alter database force logging;  

Database altered.

SQL> alter database add supplemental log data;  

Database altered.

SQL> select LOG_MODE,FORCE_LOGGING,SUPPLEMENTAL_LOG_DATA_MIN
    from v$database;

LOG_MODE     FOR SUPPLEME
------------ --- --------
ARCHIVELOG   YES YES

1.4.    准备测试的表

SQL> conn scott/tiger
Connected.
SQL> create table test(id int,name varchar2(),mark varchar2());

Table created.

SQL> select * from tab;

TNAME                          TABTYPE  CLUSTERID
------------------------------ ------- ----------
BONUS                          TABLE
DEPT                           TABLE
EMP                            TABLE
GTT                            TABLE
SALGRADE                       TABLE
TEMP                           TABLE
TEST                           TABLE

 rows selected.

2. 创建发布者和订阅者

2.1.    创建发布者/授权

SQL> create tablespace cdc_tbsp
datafile '/u02/app/oradata/ORCL/cdc_tbsp01.dbf' size 200m;

Tablespace created.

SQL> create user cdc_publisher identified by cdc_publisher
default tablespace cdc_tbsp temporary tablespace temp;

User created.

SQL> grant create session TO cdc_publisher;

Grant succeeded.

SQL> grant create table TO cdc_publisher;

Grant succeeded.

SQL> grant create sequence TO cdc_publisher; 

Grant succeeded.

SQL> grant create procedure TO cdc_publisher;  

Grant succeeded.

SQL> grant create any job TO cdc_publisher;  

Grant succeeded.

SQL> grant execute_catalog_role TO cdc_publisher; 

Grant succeeded.

SQL> grant select_catalog_role TO cdc_publisher;  

Grant succeeded.

SQL> grant execute ON dbms_cdc_publish TO cdc_publisher;  

Grant succeeded.

SQL> grant execute ON dbms_lock TO cdc_publisher;  

Grant succeeded.

SQL> GRANT UNLIMITED TABLESPACE TO cdc_publisher;

Grant succeeded.

SQL> execute dbms_streams_auth.grant_admin_privilege('CDC_PUBLISHER');  

PL/SQL procedure successfully completed.

SQL> grant all on scott.test to cdc_publisher;

Grant succeeded.

2.2.    创建订阅者/授权

SQL> create user cdc_subscriber identified by cdc_subscriber
    default tablespace cdc_tbsp temporary tablespace temp;

User created.

SQL> grant create session TO cdc_subscriber;

Grant succeeded.

3. 发布/订阅具体数据

3.1.    发布：准备源表(Source Table)

SQL> conn cdc_publisher/cdc_publisher
Connected.
SQL> BEGIN
    DBMS_CAPTURE_ADM.PREPARE_TABLE_INSTANTIATION(
TABLE_NAME => 'scott.test');
    END;
    /

PL/SQL procedure successfully completed.

3.2.    发布：创建变更集(Data Set)

SQL> conn cdc_publisher/cdc_publisher
Connected.
SQL> BEGIN
    DBMS_CDC_PUBLISH.CREATE_CHANGE_SET(
    change_set_name => 'CDC_SCOTT_TEST',
    description => 'Change set for product info',
    change_source_name => 'HOTLOG_SOURCE',
    stop_on_ddl => 'y',
    begin_date => sysdate,
    end_date => sysdate+);
    END;
   /

PL/SQL procedure successfully completed.

3.3.    发布：创建变更表

SQL> BEGIN
       DBMS_CDC_PUBLISH.CREATE_CHANGE_TABLE(
       owner              => 'cdc_publisher',
       change_table_name  => 'cdc_test',
       change_set_name    => 'CDC_SCOTT_TEST',
       source_schema      => 'SCOTT',
       source_table       => 'TEST',
       column_type_list   => 'ID NUMBER(5), NAME VARCHAR2(30),MARK VARCHAR2(50)',
       capture_values     => 'both',
      rs_id              => 'y',
      row_id             => 'n',
      user_id            => 'n',
      timestamp          => 'n',
      object_id          => 'n',
      source_colmap      => 'n',
      target_colmap      => 'y',
      options_string     => 'TABLESPACE CDC_TBSP');
   END;
   /

PL/SQL procedure successfully completed.

3.4.    发布：激活变更集

SQL> BEGIN
       DBMS_CDC_PUBLISH.ALTER_CHANGE_SET(
          change_set_name => 'CDC_SCOTT_TEST',
          enable_capture => 'y');
    END;
    /

PL/SQL procedure successfully completed.

SQL> grant select on  cdc_test to cdc_subscriber;

Grant succeeded.

备注：其实到此cdc_subscriber用户已经可以检测到scott.test表的变更了

如下测试：

[oracle@std ~]$ sqlplus scott/tiger

SQL*Plus: Release 11.2.0.1. Production on Wed Jan  :: 

Copyright (c) , , Oracle.  All rights reserved.

Connected to:
Oracle Database 11g Enterprise Edition Release 11.2.0.1. - Production
With the Partitioning, OLAP, Data Mining and Real Application Testing options

SQL> insert into test values(,'beijing','');

 row created.

SQL> commit;

Commit complete.

SQL> update test set name='shanghai' where id=;

 row updated.

SQL> commit;

Commit complete.

SQL> delete test where id=;

 row deleted.

SQL> commit;

Commit complete.

SQL> conn cdc_subscriber/cdc_subscriber
Connected.
SQL> select t.operation$,t.commit_timestamp$,t.id,t.name,t.mark
    from cdc_publisher.cdc_test t;

OP COMMIT_TIMESTAMP$          ID NAME                           MARK
-- ------------------ ---------- ------------------------------
I  -JAN-                    beijing
UO -JAN-                    beijing
UN -JAN-                    shanghai
D  -JAN-                    shanghai                       

3.5.    订阅：创建订阅集

SQL> conn cdc_subscriber/cdc_subscriber
Connected.
SQL> BEGIN
    dbms_cdc_subscribe.create_subscription(
    change_set_name=>'CDC_SCOTT_TEST',
    description=>'cdc scott subx',
    subscription_name=>'CDC_SCOTT_SUB');
    END;
    /

PL/SQL procedure successfully completed.

3.6.    订阅：开始订阅表信息

SQL> BEGIN
    dbms_cdc_subscribe.subscribe(
    subscription_name=>'CDC_SCOTT_SUB',
    source_schema=>'SCOTT',
    source_table=>'TEST',
    column_list=>'ID, NAME,MARK',
    subscriber_view=>'TEST_TEMP');
    END;
    /

PL/SQL procedure successfully completed.

SQL> select * from tab;

TNAME                          TABTYPE  CLUSTERID
------------------------------ ------- ----------
TEST_TEMP                      VIEW

3.7.    订阅：激活订阅

SQL> BEGIN
    dbms_cdc_subscribe.activate_subscription(
    subscription_name=>'CDC_SCOTT_SUB');
    END;
    /

PL/SQL procedure successfully completed.

3.8.    订阅：扩展订阅窗口

SQL> conn cdc_subscriber/cdc_subscriber
Connected.
SQL> BEGIN
    dbms_cdc_subscribe.extend_window(
    subscription_name=>'CDC_SCOTT_SUB');
    END;
    /

PL/SQL procedure successfully completed.

SQL> conn cdc_subscriber/cdc_subscriber
Connected.
SQL> select t.operation$,t.commit_timestamp$,t.id,t.name,t.mark
    from test_temp t; 

OP COMMIT_TIMESTAMP$          ID NAME                           MARK
-- ------------------ ---------- ------------------------------
I  -JAN-                    beijing
UO -JAN-                    beijing
UN -JAN-                    shanghai
D  -JAN-                    shanghai                       

4. 测试订阅发布

4.1.    SCOTT表更改

SQL> conn scott/tiger
Connected.
SQL> insert into test values(,'renqinglei','aa');

 row created.

SQL> commit;

Commit complete.

SQL> update test set mark='tt' where id=;

 row updated.

SQL> commit;

Commit complete.

SQL> delete test where id=;

 row deleted.

SQL> commit;

Commit complete.

4.2.    查询数据发布情况

SQL> conn cdc_publisher/cdc_publisher
Connected.
SQL> select t.operation$,t.commit_timestamp$,t.id,t.name,t.mark
    from cdc_publisher.cdc_test t;

OP COMMIT_TIMESTAMP$          ID NAME                           MARK
-- ------------------ ---------- ------------------------------
I  -JAN-                    beijing
UO -JAN-                    beijing
UN -JAN-                    shanghai
D  -JAN-                    shanghai
I  -JAN-                    renqinglei                     aa
UO -JAN-                    renqinglei                     aa
UN -JAN-                    renqinglei                     tt
D  -JAN-                    renqinglei                     tt

 rows selected.

4.3.    查询数据订阅情况

SQL> conn cdc_subscriber/cdc_subscriber
Connected.
SQL> select t.operation$,t.commit_timestamp$,t.id,t.name,t.mark
    from test_temp t; 

OP COMMIT_TIMESTAMP$          ID NAME                           MARK
-- ------------------ ---------- ------------------------------
I  -JAN-                    beijing
UO -JAN-                    beijing
UN -JAN-                    shanghai
D  -JAN-                    shanghai                       

发现订阅的数据没有变化,扩展一下订阅窗口:

SQL> conn cdc_subscriber/cdc_subscriber
Connected.
SQL> BEGIN
    dbms_cdc_subscribe.extend_window(
    subscription_name=>'CDC_SCOTT_SUB');
    END;
    /

PL/SQL procedure successfully completed.

SQL> select t.operation$,t.commit_timestamp$,t.id,t.name,t.mark
    from test_temp t; 

OP COMMIT_TIMESTAMP$          ID NAME                           MARK
-- ------------------ ---------- ------------------------------
I  -JAN-                    beijing
UO -JAN-                    beijing
UN -JAN-                    shanghai
D  -JAN-                    shanghai
I  -JAN-                    renqinglei                     aa
UO -JAN-                    renqinglei                     aa
UN -JAN-                    renqinglei                     tt
D  -JAN-                    renqinglei                     tt

 rows selected.

4.4.    清除变更数据集

SQL> conn cdc_subscriber/cdc_subscriber
Connected.
SQL> BEGIN
    DBMS_CDC_SUBSCRIBE.PURGE_WINDOW(
    subscription_name => 'CDC_SCOTT_SUB');
    END;
    /

PL/SQL procedure successfully completed.

SQL> select t.operation$,t.commit_timestamp$,t.id,t.name,t.mark
    from test_temp t;

no rows selected

4.5.    重新生成变化数据

SQL> conn scott/tiger
Connected.
SQL> insert into test values(,'shandong','hh');

 row created.

SQL> insert into test values(,'diankeyuan','hh');

 row created.

SQL> commit;

Commit complete.

SQL> conn cdc_publisher/cdc_publisher
Connected.
SQL> select t.operation$,t.commit_timestamp$,t.id,t.name,t.mark
    from cdc_publisher.cdc_test t;

OP COMMIT_TIMESTAMP$          ID NAME                           MARK
-- ------------------ ---------- ------------------------------
I  -JAN-                    beijing
UO -JAN-                    beijing
UN -JAN-                    shanghai
D  -JAN-                    shanghai
I  -JAN-                    renqinglei                     aa
UO -JAN-                    renqinglei                     aa
UN -JAN-                    renqinglei                     tt
D  -JAN-                    renqinglei                     tt
I  -JAN-                    shandong                       hh
I  -JAN-                    diankeyuan                     hh

 rows selected.

SQL> conn cdc_subscriber/cdc_subscriber
Connected.
SQL> BEGIN
    dbms_cdc_subscribe.extend_window(
    subscription_name=>'CDC_SCOTT_SUB');
    END;
    /

PL/SQL procedure successfully completed.

SQL> select t.operation$,t.commit_timestamp$,t.id,t.name,t.mark
    from test_temp t;

OP COMMIT_TIMESTAMP$          ID NAME                           MARK
-- ------------------ ---------- ------------------------------
I  -JAN-                    shandong                       hh
I  -JAN-                    diankeyuan                     hh

4.6.    删除发布的数据

SQL> conn cdc_publisher/cdc_publisher
Connected.
SQL> select t.operation$,t.commit_timestamp$,t.id,t.name,t.mark
    from cdc_publisher.cdc_test t;

OP COMMIT_TIMESTAMP$          ID NAME                           MARK
-- ------------------ ---------- ------------------------------
I  -JAN-                    beijing
UO -JAN-                    beijing
UN -JAN-                    shanghai
D  -JAN-                    shanghai
I  -JAN-                    renqinglei                     aa
UO -JAN-                    renqinglei                     aa
UN -JAN-                    renqinglei                     tt
D  -JAN-                    renqinglei                     tt
I  -JAN-                    shandong                       hh
I  -JAN-                    diankeyuan                     hh

 rows selected.

SQL> truncate cdc_test;
truncate cdc_test
                *
ERROR at line :
ORA-: Invalid truncate command - missing CLUSTER or TABLE keyword

SQL> delete cdc_test;

 rows deleted.

SQL> commit;

Commit complete.

SQL> select t.operation$,t.commit_timestamp$,t.id,t.name,t.mark
    from cdc_publisher.cdc_test t;

no rows selected