環境說明

本實驗環境基於Oracle 12C和JDK1.8，其中Oracle 12C支援多租戶特性，相較於之前的Oracle版本，使用‘C##使用者名稱‘表示使用者，例如如果資料庫使用者叫kevin，則登陸時使用C##kevin進行登陸。

一、Oracle高階訊息佇列AQ

Oracle AQ是Oracle中的訊息佇列，是Oracle中的一種高階應用，每個版本都在不斷的加強，使用DBMS_AQ系統包進行相應的操作，是Oracle的預設元件，只要安裝了Oracle資料庫就可以使用。使用AQ可以在多個Oracle資料庫、Oracle與Java、C等系統中進行資料傳輸。

下面分步驟說明如何建立Oracle AQ

1. 建立訊息負荷payload

Oracle AQ中傳遞的訊息被稱為有效負荷（payloads），格式可以是使用者自定義物件或XMLType或ANYDATA。本例中我們建立一個簡單的物件型別用於傳遞訊息。

create type demo_queue_payload_type as object (message varchar2(4000));

2. 建立隊列表

隊列表用於儲存訊息，在入隊時自動存入表中，出隊時自動刪除。使用DBMS_AQADM包進行資料表的建立，只需要寫表名，同時設定相應的屬性。對於佇列需要設定multiple_consumers為false，如果使用釋出/訂閱模式需要設定為true。

begin
  dbms_aqadm.create_queue_table(
    queue_table   => 'demo_queue_table',
    queue_payload_type => 'demo_queue_payload_type',
    multiple_consumers => false
  );
end;

執行完後可以檢視oracle表中自動生成了demo_queue_table表，可以檢視影響子段（含義比較清晰）。

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 != 'DEMO_QUEUE_PAYLOAD_TYPE';

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

我們看到一個佇列帶出了一系列自動生成物件，有些是被後面直接用到的。不過有趣的是，建立了第二個佇列。這就是所謂的異常佇列（exception queue）。如果AQ無法從我們的佇列接收訊息，將記錄在該異常佇列中。

訊息多次處理出錯等情況會自動轉移到異常的佇列，對於異常佇列如何處理目前筆者還沒有找到相應的寫法，因為我使用的場景並不要求訊息必須一對一的被處理，只要起到通知的作用即可。所以如果訊息轉移到異常佇列，可以執行清空隊列表中的資料

delete from demo_queue_table;

4. 佇列的停止和刪除

如果需要刪除或重建可以使用下面的方法進行操作：

BEGIN
   DBMS_AQADM.STOP_QUEUE(
      queue_name => 'demo_queue'
      );
   DBMS_AQADM.DROP_QUEUE(
      queue_name => 'demo_queue'
      );
   DBMS_AQADM.DROP_QUEUE_TABLE(
      queue_table => 'demo_queue_table'
      );
END;

5. 入隊訊息

入列操作是一個基本的事務操作（就像往隊列表Insert），因此我們需要提交。

declare
  r_enqueue_options DBMS_AQ.ENQUEUE_OPTIONS_T;
  r_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
  v_message_handle RAW(16);
  o_payload demo_queue_payload_type;
begin
  o_payload := demo_queue_payload_type('what is you name ?');

  dbms_aq.enqueue(
    queue_name  => 'demo_queue',
    enqueue_options => r_enqueue_options,
    message_properties => r_message_properties,
    payload => o_payload,
    msgid => v_message_handle
  );

  commit;
end;

通過SQL語句檢視訊息是否正常入隊：

select * from aq$demo_queue_table;
select user_data from aq$demo_queue_table;

6. 出隊訊息

使用Oracle進行出隊操作

declare
  r_dequeue_options DBMS_AQ.DEQUEUE_OPTIONS_T;
  r_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
  v_message_handle RAW(16);
  o_payload demo_queue_payload_type;
begin
  DBMS_AQ.DEQUEUE(
    queue_name => 'demo_queue',
    dequeue_options => r_dequeue_options,
    message_properties => r_message_properties,
    payload => o_payload,
    msgid => v_message_handle
  );


  DBMS_OUTPUT.PUT_LINE(
    '***** Browse message is [' || o_payload.message || ']****'
  );

end;

二、Java使用JMS監聽並處理Oracle AQ佇列

Java使用JMS進行相應的處理，需要使用Oracle提供的jar，在Oracle安裝目錄可以找到：在linux中可以使用find命令進行查詢，例如

find `pwd` -name 'jmscommon.jar'

需要的jar為：

• app/oracle/product/12.1.0/dbhome_1/rdbms/jlib/jmscommon.jar
• app/oracle/product/12.1.0/dbhome_1/jdbc/lib/ojdbc7.jar
• app/oracle/product/12.1.0/dbhome_1/jlib/orai18n.jar
• app/oracle/product/12.1.0/dbhome_1/jlib/jta.jar
• app/oracle/product/12.1.0/dbhome_1/rdbms/jlib/aqapi_g.jar

1. 建立連線引數類

實際使用時可以把引數資訊配置在properties檔案中，使用Spring進行注入。

package org.kevin.jms;
/**
 * 
 *  連線引數資訊
 *
 */
public class JmsConfig {

    public String username = "c##kevin";
    public String password = "a111111111";
    public String jdbcUrl = "jdbc:oracle:thin:@127.0.0.1:1521:orcl";
    public String queueName = "demo_queue";
}

2. 建立訊息轉換類

因為訊息載荷是Oracle資料型別，需要提供一個轉換工廠類將Oracle型別轉換為Java型別。

package org.kevin.jms;

import java.sql.SQLException;

import oracle.jdbc.driver.OracleConnection;
import oracle.jdbc.internal.OracleTypes;
import oracle.jpub.runtime.MutableStruct;
import oracle.sql.CustomDatum;
import oracle.sql.CustomDatumFactory;
import oracle.sql.Datum;
import oracle.sql.STRUCT;

/**
 * 
 * 資料型別轉換類
 *
 */
@SuppressWarnings("deprecation")
public class QUEUE_MESSAGE_TYPE implements CustomDatum, CustomDatumFactory {
    public static final String _SQL_NAME = "QUEUE_MESSAGE_TYPE";
    public static final int _SQL_TYPECODE = OracleTypes.STRUCT;

    MutableStruct _struct;
    // 12表示字串
    static int[] _sqlType = { 12 };
    static CustomDatumFactory[] _factory = new CustomDatumFactory[1];
    static final QUEUE_MESSAGE_TYPE _MessageFactory = new QUEUE_MESSAGE_TYPE();

    public static CustomDatumFactory getFactory() {
        return _MessageFactory;
    }

    public QUEUE_MESSAGE_TYPE() {
        _struct = new MutableStruct(new Object[1], _sqlType, _factory);
    }

    public Datum toDatum(OracleConnection c) throws SQLException {
        return _struct.toDatum(c, _SQL_NAME);
    }

    public CustomDatum create(Datum d, int sqlType) throws SQLException {
        if (d == null)
            return null;
        QUEUE_MESSAGE_TYPE o = new QUEUE_MESSAGE_TYPE();
        o._struct = new MutableStruct((STRUCT) d, _sqlType, _factory);
        return o;
    }

    public String getContent() throws SQLException {
        return (String) _struct.getAttribute(0);
    }

}

3. 主類進行訊息處理

package org.kevin.jms;

import java.util.Properties;

import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.Session;

import oracle.jms.AQjmsAdtMessage;
import oracle.jms.AQjmsDestination;
import oracle.jms.AQjmsFactory;
import oracle.jms.AQjmsSession;

/**
 * 
 * 訊息處理類
 *
 */
public class Main {

    public static void main(String[] args) throws Exception {
        JmsConfig config = new JmsConfig();

        QueueConnectionFactory queueConnectionFactory = AQjmsFactory.getQueueConnectionFactory(config.jdbcUrl,
                new Properties());

        QueueConnection conn = queueConnectionFactory.createQueueConnection(config.username, config.password);
        AQjmsSession session = (AQjmsSession) conn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

        conn.start();

        Queue queue = (AQjmsDestination) session.getQueue(config.username, config.queueName);
        MessageConsumer consumer = session.createConsumer(queue, null, QUEUE_MESSAGE_TYPE.getFactory(), null, false);

        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                System.out.println("ok");

                AQjmsAdtMessage adtMessage = (AQjmsAdtMessage) message;

                try {
                    QUEUE_MESSAGE_TYPE payload = (QUEUE_MESSAGE_TYPE) adtMessage.getAdtPayload();
                    System.out.println(payload.getContent());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        Thread.sleep(1000000);
    }

}

使用Oracle程式塊進行入隊操作，在沒有啟動Java時看到隊列表中存在資料。啟動Java後，控制檯正確的輸出的訊息；通過Oracle程式塊再次寫入訊息，發現控制檯正確處理訊息。Java的JMS監聽不是立刻進行處理，可能存在幾秒中的時間差，時間不等。

三、監控表記錄變化通知Java

下面的例子建立一個數據表，然後在表中新增觸發器，當資料變化後觸發器呼叫儲存過程給Oracle AQ傳送訊息，然後使用Java JMS對訊息進行處理。

1. 建立表

建立student表，包含username和age兩個子段，其中username時varchar2型別，age時number型別。

2. 建立儲存過程

建立send_aq_msg儲存過程，因為儲存過程中呼叫dbms資料包，系統包在儲存過程中執行需要進行授權（使用sys使用者進行授權）：

grant execute on dbms_aq to c##kevin;

• 1

注意儲存過程中包含commit語句。

create or replace 
PROCEDURE send_aq_msg (info IN VARCHAR2) as
  r_enqueue_options DBMS_AQ.ENQUEUE_OPTIONS_T;
  r_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
  v_message_handle RAW(16);
  o_payload demo_queue_payload_type;
begin
  o_payload := demo_queue_payload_type(info);

  dbms_aq.enqueue(
    queue_name  => 'demo_queue',
    enqueue_options => r_enqueue_options,
    message_properties => r_message_properties,
    payload => o_payload,
    msgid => v_message_handle
  );

  commit;
end send_aq_msg;

3. 建立觸發器

在student表中建立觸發器，當資料寫入或更新時，如果age=18，則進行入隊操作。需要呼叫儲存過程傳送訊息，但觸發器中不能包含事物提交語句，因此需要使用pragma autonomous_transaction;宣告自由事物：

CREATE OR REPLACE TRIGGER STUDENT_TR 
AFTER INSERT OR UPDATE OF AGE ON STUDENT FOR EACH ROW 
DECLARE
pragma autonomous_transaction;
BEGIN
  if :new.age = 18 then
      send_aq_msg(:new.username);  
  end if;  
END;