oracle 11g streams 传播进程使用示例

--创建存储LCR队列
conn strmadmin/strmadmin@DBXA.WORLD
Connected.
begin
    dbms_streams_adm.set_up_queue(
        queue_table => 'DBXA_CAP_Q_T',
        storage_clause => 'TABLESPACE STREAMS_TBS',
        queue_name => 'DBXA_CAP_Q',
        queue_user => 'STRMADMIN',
        comment => 'Queue for Capture Process'
    );
end;
/
PL/SQL procedure successfully completed.

--查看上面存储过程Strmadmin schema下创建的对象
select table_name,
    tablespace_name,
    iot_type
from tabs
where table_name like '%DBXA%';
TABLE_NAME           TABLESPACE_NAME IOT_TYPE
-------------------- --------------- ------------
DBXA_CAP_Q_T         STREAMS_TBS
AQ$_DBXA_CAP_Q_T_S   STREAMS_TBS
AQ$_DBXA_CAP_Q_T_L   STREAMS_TBS
AQ$_DBXA_CAP_Q_T_P   STREAMS_TBS
AQ$_DBXA_CAP_Q_T_C                   IOT
AQ$_DBXA_CAP_Q_T_D                   IOT
AQ$_DBXA_CAP_Q_T_G                   IOT
AQ$_DBXA_CAP_Q_T_H                   IOT
AQ$_DBXA_CAP_Q_T_I                   IOT
AQ$_DBXA_CAP_Q_T_T                   IOT

10 rows selected.

--上面存储过程还创建了 缓冲区队列、exception 队列
select name,
    queue_table,
    queue_type,
    user_comment
from dba_queues
where name like '%DBXA_CAP_Q%';
NAME                QUEUE_TABLE   QUEUE_TYPE       USER_COMMENT
------------------- ------------- ---------------- -------------------------
DBXA_CAP_Q          DBXA_CAP_Q_T  NORMAL_QUEUE     Queue for Capture Process
AQ$_DBXA_CAP_Q_T_E  DBXA_CAP_Q_T  EXCEPTION_QUEUE  exception queue

--查看上面创建队列表DBXA_CAP_Q_T相关联的队列
select queue_table,
    object_type,            --相关联队列类型
    sort_order,             --队列是怎样存储消息的（COMMIT_TIME：按照提交时的顺序）
    message_grouping,       --消息组合方式（transactional：事务性）
    secure                  --是否是安全队列
from dba_queue_tables
where owner = 'STRMADMIN';
QUEUE_TABLE   OBJECT_TYPE  SORT_ORDER   MESSAGE_GROUP SEC
------------- ------------ ------------ ------------- ---
DBXA_CAP_Q_T  SYS.ANYDATA  COMMIT_TIME  TRANSACTIONAL YES

--定义队列的 primary instance 和 secondary instance
--下面显示了 instance 2 为primary，instance 3 为secondary
begin
    dbms_aqadm.alter_queue_table (
        queue_table => 'DBXA_CAP_Q_T',
        primary_instance => 2,
        secondary_instance => 3
    );
end;
/
PL/SQL procedure successfully completed.

--查看当前队列拥有者实例
select queue_table,
    owner_instance,
    primary_instance,
    secondary_instance
from dba_queue_tables
where owner = 'STRMADMIN';
QUEUE_TABLE     OWNER_INSTANCE PRIMARY_INSTANCE SECONDARY_INSTANCE
--------------- -------------- ---------------- ------------------
DBXA_CAP_Q_T                 2                2                  3

declare
    l_dml_rule_name varchar2(30);
    l_ddl_rule_name varchar2(30);
begin
    dbms_streams_adm.add_schema_propagation_rules (
        schema_name => 'SCOTT',
        streams_name => 'DBXA_TO_DBXB_PROP',
        source_queue_name => 'DBXA_CAP_Q',
        destination_queue_name => 'DBXA_APP_Q@DBXB.WORLD',
        include_dml => true,
        include_ddl => true,
        source_database => 'DBXA.WORLD',
        dml_rule_name => l_dml_rule_name,
        ddl_rule_name => l_ddl_rule_name,
        inclusion_rule => true,
        queue_to_queue => true,
        and_condition => ' :lcr.get_compatible() < 
                dbms_streams.max_compatible()'
    );
    dbms_output.put_line('DML Rule Name is : ' || l_dml_rule_name);
    dbms_output.put_line('DDL Rule Name is : ' || l_ddl_rule_name);
end;
/
DML Rule Name is : "STRMADMIN"."SCOTT153"
DDL Rule Name is : "STRMADMIN"."SCOTT154"
PL/SQL procedure successfully completed.
PL/SQL procedure successfully completed.

--查看上面自动创建的传播进程相关信息
select rule_set_name,
    source_queue_name,
    destination_queue_name dest_queue_name,
    destination_dblink dest_dblink
from dba_propagation
where propagation_name = 'DBXA_TO_DBXB_PROP';
RULE_SET_NAME   SOURCE_QUEUE_NAME    DEST_QUEUE_NAME  DEST_DBLINK
--------------- -------------------- ---------------- ---------------
RULESET$_155    DBXA_CAP_Q           DBXA_APP_Q       DBXB.WORLD

--查看上面自动创建的传播进程的其他信息
select status,
    source_queue_owner src_owner,
    source_queue_name src_q_name,
    destination_queue_owner dest_owner,
    destination_queue_name dest_q_name,
    destination_dblink dest_dblink,
    queue_to_queue q_to_q
from dba_propagation
where propagation_name = 'DBXA_TO_DBXB_PROP';
STATUS   SRC_OWNER  SRC_Q_NAME  DEST_OWNER DEST_Q_NAME  DEST_DBLINK Q_TO_Q
-------- ---------- ----------- ---------- ------------ ----------- ------
ENABLED  STRMADMIN  DBXA_CAP_Q  STRMADMIN  DBXA_APP_Q   DBXB.WORLD  TRUE

begin
    dbms_propagation_adm.create_propagation (
        propagation_name => 'DBXA_TO_DBXB_PROP',
        source_queue => 'DBXA_CAP_Q',
        destination_queue => 'DBXA_APP_Q',
        destination_dblink => 'DBXB',
        rule_set_name => 'RULESET$_155',
        queue_to_queue => TRUE
    );
end;
/
PL/SQL procedure successfully completed.

--查看传播进程相关信息
select status,
    source_queue_owner src_owner,
    source_queue_name src_q_name,
    destination_queue_owner dest_owner,
    destination_queue_name dest_q_name,
    destination_dblink dest_dblink,
    queue_to_queue q_to_q
from dba_propagation
where propagation_name = 'DBXA_TO_DBXB_PROP';
STATUS   SRC_OWNER  SRC_Q_NAME  DEST_OWNER DEST_Q_NAME  DEST_DBLINK Q_TO_Q
-------- ---------- ----------- ---------- ------------ ----------- ------
ENABLED  STRMADMIN  DBXA_CAP_Q  STRMADMIN  DBXA_APP_Q   DBXB.WORLD  TRUE

--移除传播进程中的 rule
begin
    dbms_propagation_adm.alter_propagation(
        propagation_name => 'DBXA_TO_DBXB_PROP',
        remove_rule_set => TRUE
    );
end;
/
PL/SQL procedure successfully completed.

--查看队列的默认调度计划
select job_name,
    message_delivery_mode delivery_mode,
    start_date,
    start_time,
    next_time,
    propagation_window window,
    latency
from dba_queue_schedules
where qname = 'DBXA_CAP_Q'
/
JOB_NAME      DELIVERY_M START_DATE  START_TIME  NEXT_TIME  WINDOW LATENCY
------------- ---------- ----------- ----------- ---------- ------ -------
AQ_JOB$_228   PERSISTENT                                                 3
AQ_JOB$_228   BUFFERED                                                   3

--查看上面job：AQ_JOB$_228的详细信息
select schedule_type sched_type,
    state,
    event_queue_name event_qname,
    event_queue_agent event_qagent,
    event_condition e_C
from dba_scheduler_jobs
where job_creator = 'STRMADMIN'
and job_name = 'AQ_JOB$_228';
SCHED_TYPE   STATE      EVENT_QNAME  EVENT_QAGENT                        E_C
------------ ---------- ------------ ----------------------------------- -----
EVENT        SCHEDULED  DBXA_CAP_Q   AQ$_P@"STRMADMIN"."DBXA_APP_Q"@DBXB (1=1)

--修改传播作业调度计划方法
SQL> begin
2 dbms_aqadm.alter_propagation_schedule(
3 queue_name => 'STRMADMIN.DBXA_CAP_Q',
4 destination => '"STRMADMIN"."DBXA_APP_Q"@DBXC.WORLD',
5 duration => 3600,
6 next_time => 'SYSDATE + 1800/86400',
7 latency => '60'
8 );
9 end;
10 /
PL/SQL procedure successfully completed

--通过视图DBA_QUEUE_SCHEDULE查看job：AQ_JOB$_228相关信息
select job_name,
    message_delivery_mode delivery_mode,
    start_date,
    start_time,
    next_time,
    propagation_window window,
    latency
from dba_queue_schedules
where JOB_NAME = 'AQ_JOB$_228';
JOB_NAME    DELIVERY_M START_DATE  START_TIME  NEXT_TIME            WINDOW LATENCY
----------- ---------- ----------- ----------- -------------------- ------ -------
AQ_JOB$_228 PERSISTENT                         SYSDATE + 1800/86400   3600      60
AQ_JOB$_228 BUFFERED                           SYSDATE + 1800/86400   3600      60

--传播作业不再基于事件
select schedule_type sched_type,
    state,
    event_queue_name event_qname,
    event_queue_agent event_qagent,
    event_condition e_C
from dba_scheduler_jobs
where job_creator = 'STRMADMIN'
and job_name = 'AQ_JOB$_228';
SCHED_TYPE   STATE      EVENT_QNAME  EVENT_QAGENT                       E_C
------------ ---------- ------------ ---------------------------------- -----
PLSQL        SCHEDULED

--查看CCA在捕获进程或应用进程中的使用
select optimization,
    capture_name,
    apply_name,
    apply_dblink
from v$streams_capture
where capture_name = 'DBXA_CAP';
CAPTURE_NAME   OPTIMIZATION APPLY_NAME  APPLY_DBLINK
-------------- ------------ ----------- --------------------
DBXA_CAP                  2 DBXA_APP    DBXB.WORLD