Oracle Advanced Queuing (AQ) 入門實戰：建立佇列、入隊/出隊、非同步通知

Oracle Advanced Queuing (AQ) 入門實戰：建立佇列、入隊/出隊、非同步通知

Oracle Advanced Queuing (AQ) 是資料庫內建的訊息佇列功能，支援 PL/SQL、OCI、JMS (透過 JMS 接適器) 等多種存取方式，提供持久化、事務性、優先級、延遲投遞、非同步通知等企業級特性。本文以 PL/SQL 為例，示範從建立佇列到同步/非同步消費的完整流程。

前置權限需求

• 角色：AQ_ADMINISTRATOR_ROLE (建立佇列表/佇列)、AQ_USER_ROLE (應用存取)
• 系統權限：EXECUTE ON DBMS_AQ, EXECUTE ON DBMS_AQADM
• 表空間配額

1. 建立佇列負載型別、佇列表、佇列

-- 1-1 定義 Payload 型別
CREATE TYPE demo_queue_payload_type AS OBJECT (message VARCHAR2(4000));
/

-- 1-2 建立佇列表 (儲存訊息的實體表)
BEGIN
  DBMS_AQADM.CREATE_QUEUE_TABLE(
    queue_table        => 'demo_queue_table',
    queue_payload_type => 'demo_queue_payload_type'
    -- multiple_consumers => TRUE  -- 非同步通知需設為 TRUE
  );
END;
/

-- 1-3 建立並啟動佇列
BEGIN
  DBMS_AQADM.CREATE_QUEUE(
    queue_name  => 'demo_queue',
    queue_table => 'demo_queue_table'
  );
  DBMS_AQADM.START_QUEUE(queue_name => 'demo_queue');
END;
/

-- 自動產生的輔助物件
SELECT object_name, object_type FROM user_objects 
WHERE object_name LIKE 'DEMO_QUEUE%' OR object_name LIKE 'AQ$_DEMO_QUEUE%';
/*
OBJECT_NAME                    OBJECT_TYPE
------------------------------ -----------
DEMO_QUEUE_TABLE               TABLE
SYS_C009392                    INDEX
SYS_LOB0000060502C00030$$      LOB
AQ$_DEMO_QUEUE_TABLE_T         INDEX
AQ$_DEMO_QUEUE_TABLE_I         INDEX
AQ$_DEMO_QUEUE_TABLE_E         QUEUE   -- 異常佇列
AQ$DEMO_QUEUE_TABLE            VIEW    -- 查詢佇列內容
DEMO_QUEUE                     QUEUE
*/

2. 同步入隊

DECLARE
  r_enq_opt   DBMS_AQ.ENQUEUE_OPTIONS_T;
  r_msg_prop  DBMS_AQ.MESSAGE_PROPERTIES_T;
  v_msgid     RAW(16);
  o_payload   demo_queue_payload_type;
BEGIN
  o_payload := demo_queue_payload_type('Here is a message');
  DBMS_AQ.ENQUEUE(
    queue_name         => 'demo_queue',
    enqueue_options    => r_enq_opt,
    message_properties => r_msg_prop,
    payload            => o_payload,
    msgid              => v_msgid
  );
  COMMIT; -- 入隊需提交事務
END;
/
-- 查詢佇列狀態
SELECT COUNT(*) FROM aq$demo_queue_table; -- 1

3. 同步瀏覽 / 出隊

瀏覽 (不移除)

DECLARE
  r_deq_opt   DBMS_AQ.DEQUEUE_OPTIONS_T;
  r_msg_prop  DBMS_AQ.MESSAGE_PROPERTIES_T;
  v_msgid     RAW(16);
  o_payload   demo_queue_payload_type;
BEGIN
  r_deq_opt.dequeue_mode := DBMS_AQ.BROWSE; -- 關鍵：瀏覽模式
  DBMS_AQ.DEQUEUE(
    queue_name         => 'demo_queue',
    dequeue_options    => r_deq_opt,
    message_properties => r_msg_prop,
    payload            => o_payload,
    msgid              => v_msgid
  );
  DBMS_OUTPUT.PUT_LINE('Browsed: ' || o_payload.message);
END;
/
-- 佇列仍有 1 則訊息

出隊 (移除)

DECLARE
  r_deq_opt   DBMS_AQ.DEQUEUE_OPTIONS_T;
  r_msg_prop  DBMS_AQ.MESSAGE_PROPERTIES_T;
  v_msgid     RAW(16);
  o_payload   demo_queue_payload_type;
BEGIN
  -- 預設 dequeue_mode = DBMS_AQ.REMOVE
  DBMS_AQ.DEQUEUE(
    queue_name         => 'demo_queue',
    dequeue_options    => r_deq_opt,
    message_properties => r_msg_prop,
    payload            => o_payload,
    msgid              => v_msgid
  );
  DBMS_OUTPUT.PUT_LINE('Dequeued: ' || o_payload.message);
  COMMIT; -- 出隊需提交
END;
/
-- 佇列變空

4. 非同步通知 — 訂閱者自動觸發 PL/SQL Callback

適用場景：訊息產生即時處理，無需輪詢。

4-1 重建支援多消費者的佇列

BEGIN
  DBMS_AQADM.STOP_QUEUE('demo_queue');
  DBMS_AQADM.DROP_QUEUE('demo_queue');
  DBMS_AQADM.DROP_QUEUE_TABLE('demo_queue_table');
END;
/

BEGIN
  DBMS_AQADM.CREATE_QUEUE_TABLE(
    queue_table        => 'demo_queue_table',
    queue_payload_type => 'demo_queue_payload_type',
    multiple_consumers => TRUE  -- 關鍵：支援多訂閱者
  );
  DBMS_AQADM.CREATE_QUEUE('demo_queue', 'demo_queue_table');
  DBMS_AQADM.START_QUEUE('demo_queue');
END;
/

-- 應用表：記錄非同步出隊結果
CREATE TABLE demo_queue_message_table (message VARCHAR2(4000));

4-2 定義 Callback 程序

CREATE OR REPLACE PROCEDURE demo_queue_callback_procedure(
  context  RAW,
  reginfo  SYS.AQ$_REG_INFO,
  descr    SYS.AQ$_DESCRIPTOR,
  payload  RAW,
  payloadl NUMBER
) AS
  r_deq_opt   DBMS_AQ.DEQUEUE_OPTIONS_T;
  r_msg_prop  DBMS_AQ.MESSAGE_PROPERTIES_T;
  v_msgid     RAW(16);
  o_payload   demo_queue_payload_type;
BEGIN
  -- 關鍵：使用描述符中的 msgid、consumer_name、queue_name 精準出隊
  r_deq_opt.msgid := descr.msg_id;
  r_deq_opt.consumer_name := descr.consumer_name;
  
  DBMS_AQ.DEQUEUE(
    queue_name         => descr.queue_name,
    dequeue_options    => r_deq_opt,
    message_properties => r_msg_prop,
    payload            => o_payload,
    msgid              => v_msgid
  );
  
  INSERT INTO demo_queue_message_table (message)
  VALUES ('Message [' || o_payload.message || '] dequeued at [' || TO_CHAR(SYSTIMESTAMP, 'DD-MON-YYYY HH24:MI:SS.FF3') || ']');
  COMMIT;
END;
/
-- 授權執行權限 (如有需要)
GRANT EXECUTE ON demo_queue_callback_procedure TO PUBLIC;

4-3 新增訂閱者並註冊通知

BEGIN
  -- 新增訂閱者
  DBMS_AQADM.ADD_SUBSCRIBER(
    queue_name => 'demo_queue',
    subscriber => SYS.AQ$_AGENT('demo_queue_subscriber', NULL, NULL)
  );
  
  -- 註冊非同步通知：PL/SQL Callback
  DBMS_AQ.REGISTER(
    SYS.AQ$_REG_INFO_LIST(
      SYS.AQ$_REG_INFO(
        'DEMO_QUEUE:DEMO_QUEUE_SUBSCRIBER',  -- 佇列:訂閱者
        DBMS_AQ.NAMESPACE_AQ,                -- 命名空間: AQ
        'plsql://DEMO_QUEUE_CALLBACK_PROCEDURE', -- Callback URL
        HEXTORAW('FF')                       -- 上下文
      )
    ),
    1  -- 註冊數量
  );
END;
/
-- 也可註冊 EMAIL (mailto://) 或 HTTP (https://) 通知

4-4 測試非同步流程

-- 入隊一則含時間戳的訊息
DECLARE
  r_enq_opt   DBMS_AQ.ENQUEUE_OPTIONS_T;
  r_msg_prop  DBMS_AQ.MESSAGE_PROPERTIES_T;
  v_msgid     RAW(16);
  o_payload   demo_queue_payload_type;
BEGIN
  o_payload := demo_queue_payload_type(TO_CHAR(SYSTIMESTAMP, 'DD-MON-YYYY HH24:MI:SS.FF3'));
  DBMS_AQ.ENQUEUE('demo_queue', r_enq_opt, r_msg_prop, o_payload, v_msgid);
  COMMIT;
END;
/

-- 稍等幾秒查詢應用表
SELECT * FROM demo_queue_message_table;
/*
MESSAGE
--------------------------------------------------------------------------------
Message [21-JUL-2005 21:54:51.156] dequeued at [21-JUL-2005 21:54:56.015]
*/
-- 可見入隊與非同步出隊約有 5 秒延遲 (輪詢間隔相關)

關鍵點總結

1. Payload 型別 可為 Object Type、XMLType、ANYDATA，滿足不同業務複雜度。
2. 事務性：入隊/出隊皆為獨立事務，需顯式 COMMIT。
3. 非同步前提：佇列表必須 MULTIPLE_CONSUMERS=TRUE，並建立 SUBSCRIBER + REGISTER。
4. Callback 參數固定：(context RAW, reginfo AQ$_REG_INFO, descr AQ$_DESCRIPTOR, payload RAW, payloadl NUMBER)，必須完全一致。
5. 異常佇列：自動建立 AQ$_QUEUE_TABLE_E，重試失敗訊息將移入此處，需定期監控清理。
6. 效能：AQ 表建議置於高效能表空間，啟用 NOLOGGING、分區、索引優化。