您现在的位置：程式碼之家 >> 數據庫知識 > >> Oracle 資料庫 >> Oracle 資料庫基礎

Oracle Advanced Queuing (AQ) 實戰：PL/SQL 實現短信推送隊列完整架構代碼

編輯：Oracle 資料庫基礎

業務場景

短信下發系統需異步處理海量下行 (MT) 與上行 (MO) 訊息，引入 AQ 實現生產者-消費者解耦、削峰填谷、可靠投遞。

AQ 核心優勢

資料庫原生特性，事務一致性、安全性、可擴展性與業務表一致。
多語言訪問：PL/SQL、Java (JMS)、C (OCI)、HTTP/HTTPS、SMTP。
支援遠程傳播 (Database Link)、異構整合 (MQSeries 透過 Messaging Gateway)。
單消費者/多消費者 (訂閱規則) 靈活模型。

實作步驟總覽

確定需求、預估效能提升。
設計訊息 Payload 物件型別 (ADT)。
隊列表/隊列管理 (DBMS_AQADM)。
入隊/出隊業務邏輯封裝 (DBMS_AQ)。

1. 授權與 ADT 定義


GRANT AQ_ADMINISTRATOR_ROLE TO smpuser;
-- 需額外授權 DBMS_AQ, DBMS_AQADM 執行權限

CREATE OR REPLACE TYPE MT_STRUC AS OBJECT (
  OP_ID NUMBER,
  DEST_MOBILENO VARCHAR2(32),
  FEE_MOBILENO VARCHAR2(32),
  LONG_NO VARCHAR2(21),
  MT_CONTENT VARCHAR2(500),
  SERVICE_ID VARCHAR2(10),
  MT_FLAG NUMBER(1),
  MSG_FMT NUMBER(1),
  FEE_TYPE NUMBER(1),
  FEE_VALUE NUMBER(4),
  BEG_REPORT NUMBER(1),
  CLIENT_ID NUMBER(4),
  CLIENT_NAME VARCHAR2(10),
  MT_TIME DATE,
  MT_ID VARCHAR2(40),
  PLAN_ID NUMBER,
  PUSH_TYPE VARCHAR2(20),
  SEND_DATE DATE
);

2. 隊列表與隊列管理 (DBMS_AQADM)

建立隊列表 (含優先級排序、事務分組、多消費者選項)


-- 基本隊列表
EXEC DBMS_AQADM.CREATE_QUEUE_TABLE(
  QUEUE_TABLE => 'SMS_MT_TAB',
  QUEUE_PAYLOAD_TYPE => 'MT_STRUC'
);

-- 進階：優先級+入隊時間排序、事務分組、多消費者
EXEC DBMS_AQADM.CREATE_QUEUE_TABLE(
  QUEUE_TABLE => 'WDZMUTIAQTABLE',
  QUEUE_PAYLOAD_TYPE => 'WDZAQMSG',
  SORT_LIST => 'PRIORITY,ENQ_TIME',
  MESSAGE_GROUPING => DBMS_AQADM.TRANSACTIONAL,
  MULTIPLE_CONSUMERS => TRUE
);

建立隊列 (正常/異常/備份)


EXEC DBMS_AQADM.CREATE_QUEUE(QUEUE_NAME => 'SMS_MT_QUEUE', QUEUE_TABLE => 'SMS_MT_TAB');
EXEC DBMS_AQADM.CREATE_QUEUE(QUEUE_NAME => 'SMS_MT_QUEUE_EXCEPTION', QUEUE_TABLE => 'SMS_MT_TAB', QUEUE_TYPE => DBMS_AQADM.EXCEPTION_QUEUE);
EXEC DBMS_AQADM.CREATE_QUEUE(QUEUE_NAME => 'SMS_MT_QUEUE_BACKUP', QUEUE_TABLE => 'SMS_MT_TAB');

啟動/停止隊列


DBMS_AQADM.START_QUEUE('SMS_MT_QUEUE', ENQUEUE => TRUE, DEQUEUE => TRUE);
DBMS_AQADM.STOP_QUEUE('SMS_MT_QUEUE', ENQUEUE => FALSE, DEQUEUE => TRUE);

3. 核心業務封裝：入隊


CREATE OR REPLACE PROCEDURE P100_MT_ENQUEUE (
  P_EQUEUE_NAME IN VARCHAR2,      -- SMS_MT_QUEUE / SMS_MT_QUEUE_BACKUP
  P_BODY IN T_OUTBOX_PUSH%ROWTYPE,
  P_LEVEL IN NUMBER DEFAULT 3,    -- 優先級 1-5 (越小越高)
  P_DELAY IN NUMBER DEFAULT 0,    -- 延遲秒數
  P_RES_STR OUT VARCHAR2,         -- 0 成功
  P_MSG_ID OUT VARCHAR2           -- 返回 MSGID
) AS
  V_ENQUEUE_OPTIONS DBMS_AQ.ENQUEUE_OPTIONS_T;
  V_MSG_PROPS DBMS_AQ.MESSAGE_PROPERTIES_T;
  V_MSG_HANDLE RAW(16);
  V_BODY MT_STRUC;
BEGIN
  P_RES_STR := '-1';
  -- 映射業務表行 → ADT 物件
  V_BODY := MT_STRUC(P_BODY.OP_ID, P_BODY.DEST_MOBILENO, ..., P_BODY.SEND_DATE);
  -- 設定屬性
  V_MSG_PROPS.EXCEPTION_QUEUE := 'SMS_MT_QUEUE_EXCEPTION';
  V_MSG_PROPS.PRIORITY := P_LEVEL;
  V_MSG_PROPS.DELAY := P_DELAY;
  -- 確保隊列啟動並入隊
  DBMS_AQADM.START_QUEUE(P_EQUEUE_NAME, TRUE, TRUE);
  DBMS_AQ.ENQUEUE(
    QUEUE_NAME => P_EQUEUE_NAME,
    ENQUEUE_OPTIONS => V_ENQUEUE_OPTIONS,
    MESSAGE_PROPERTIES => V_MSG_PROPS,
    PAYLOAD => V_BODY,
    MSGID => V_MSG_HANDLE
  );
  P_MSG_ID := RAWTOHEX(V_MSG_HANDLE);
  COMMIT;
  P_RES_STR := '0';
EXCEPTION
  WHEN OTHERS THEN
    ROLLBACK;
    P_RES_STR := '-100';
    -- 記錄錯誤日誌略
END;

4. 核心業務封裝：出隊 (供 Java 消費端調用)


CREATE OR REPLACE PROCEDURE P102_MT_DEQUEUE (
  P_EQUEUE_NAME IN VARCHAR2,
  P_CLIENT_ID OUT NUMBER,
  P_RES_STR OUT VARCHAR2,
  P_LABEL OUT VARCHAR2,   -- 組裝給網關的標籤
  P_BODY OUT VARCHAR2     -- 組裝給網關的報文體
) AS
  V_DEQ_OPT DBMS_AQ.DEQUEUE_OPTIONS_T;
  V_MSG_PROPS DBMS_AQ.MESSAGE_PROPERTIES_T;
  V_MSG_HANDLE RAW(16);
  V_BODY_QUEUE MT_STRUC;
  V_RES VARCHAR2(20);
  V_CNT NUMBER;
BEGIN
  P_RES_STR := '-1';
  -- 檢查隊列表是否有訊息 (簡易判斷)
  SELECT COUNT(0) INTO V_CNT FROM SMS_MT_TAB WHERE Q_NAME = P_EQUEUE_NAME;
  IF V_CNT > 0 THEN
    DBMS_AQADM.START_QUEUE(P_EQUEUE_NAME, TRUE, TRUE);
    DBMS_AQ.DEQUEUE(
      QUEUE_NAME => P_EQUEUE_NAME,
      DEQUEUE_OPTIONS => V_DEQ_OPT,
      MESSAGE_PROPERTIES => V_MSG_PROPS,
      PAYLOAD => V_BODY_QUEUE,
      MSGID => V_MSG_HANDLE
    );
    -- 將 ADT 轉為網關協議字串
    P103_CHANGE_LABEL_BODY_STR(V_BODY_QUEUE, V_RES, P_LABEL, P_BODY);
    IF V_RES = '0' THEN
      P_CLIENT_ID := V_BODY_QUEUE.CLIENT_ID;
      COMMIT;
      P_RES_STR := '0';
    ELSE
      P_RES_STR := '出隊後組串失敗';
    END IF;
  ELSE
    P_RES_STR := '隊列為空，稍後重試';
  END IF;
EXCEPTION
  WHEN OTHERS THEN ROLLBACK; P_RES_STR := '-100'; -- 記錄日誌
END;

5. 協議轉換輔助過程 (ADT → 網關 Label/Body)


CREATE OR REPLACE PROCEDURE P103_CHANGE_LABEL_BODY_STR (
  P_DEQUEUE_BODY IN MT_STRUC,
  P_RES_STR OUT VARCHAR2,
  P_LABEL OUT VARCHAR2,
  P_BODY OUT VARCHAR2
) AS
  V_LABEL VARCHAR2(50);
  V_BODY_STR VARCHAR2(2000);
BEGIN
  -- Label: 命令類型+運營商+地區+時間戳
  SELECT TO_CHAR(SYSDATE,'YY-MM-DD HH24:MI:SS') INTO V_LABEL FROM DUAL;
  IF P_DEQUEUE_BODY.CLIENT_ID > 150 THEN
    V_LABEL := '022000000' || V_LABEL;
  ELSE
    V_LABEL := '021000000' || V_LABEL;
  END IF;
  -- Body: 固定長度欄位拼接 (LPAD 對齊)
  V_BODY_STR := '0000' || '0000' || '0' || LPAD(' ',8) || LPAD(' ',16) || ... || P_DEQUEUE_BODY.MT_CONTENT;
  P_LABEL := V_LABEL; P_BODY := V_BODY_STR; P_RES_STR := '0';
EXCEPTION WHEN OTHERS THEN P_RES_STR := SQLERRM; END;