Oracle Database 21cで使える高性能キューイング「Transactional Event Queues (TEQ)」を使ってみた

## Transactional Event Queues (TEQ) とは

主な特徴は表としてキューが管理されていること、それに伴いOracle Databaseが従来から持っているデータ整合性やパフォーマンス、可用性の特徴がキューイングに適用できることです。
またKafka Javaクライアントが利用できるため、Kafka向けに作られた仕組みをTEQに置き換えられます。

その他のポイントも含め、マニュアルからいくつか抽出してみました。
[Transactional Event QueuesおよびAdvanced Queuingの概要](https://docs.oracle.com/cd/F39414_01/adque/aq-introduction.html#GUID-95868022-ECDA-4685-9D0A-52ED7663C84B)

+ Oracle Transactional Event Queues (TEQ)は、データベース統合型のメッセージ・キューイング機能
+ キューごとに複数のイベント・ストリームを保持する高性能のパーティション化された実装
+ TEQはデータベース表に実装されるため、運用上の利点である高可用性、スケーラビリティ、および信頼性のすべてがキュー・データにも当てはまる
+ 標準のデータベース機能(リカバリ、再起動、セキュリティなど)がサポート
+ のデータベース表と同様に、キュー表もインポートおよびエクスポート可能
+ 高可用性に対応するためにOracle Data Guardをサポート
+ エンキューおよびデキューは、データベース内にあることにより、分散トランザクションを必要とせずにデータベース・トランザクションに組み入れ可能
+ 標準のSQLを使用してメッセージを問い合せ可能
+ 利用可能なすべてのSQLテクノロジ(インメモリー・ラッチや表索引など)が使用可能

## TEQに対してPL/SQL経由でエンキュー、デキューしてみた

まずはTEQ用のユーザを作成し、権限を付与します。

```sql
-- ユーザ作成
CREATE USER tequsr IDENTIFIED BY tequsr ;

-- 権限付与
GRANT CREATE SESSION TO tequsr;
GRANT RESOURCE TO tequsr;
GRANT UNLIMITED TABLESPACE TO tequsr;
GRANT AQ_USER_ROLE TO tequsr;
GRANT EXECUTE ON DBMS_AQ TO tequsr;
GRANT EXECUTE ON DBMS_AQADM TO tequsr;
GRANT EXECUTE ON DBMS_AQIN TO tequsr;
```

作成したユーザでDBログインし、TEQを作成します。
今回はDBオブジェクトを格納するキューを作成します。
プロシージャの各引数については下記マニュアルを参照ください。
[CREATE_TRANSACTIONAL_EVENT_QUEUEプロシージャ](https://docs.oracle.com/cd/F39414_01/arpls/DBMS_AQADM.html#GUID-6841A667-1021-4E5C-8567-F71913AA4773)

```sql
CREATE type Message_type as object (subject VARCHAR2(30), text VARCHAR2(80));
/

BEGIN
 DBMS_AQADM.CREATE_TRANSACTIONAL_EVENT_QUEUE(
     queue_name         =>'objType_TEQ',
     storage_clause     =>null,
     multiple_consumers =>true,
     max_retries        =>10,
     comment            =>'ObjectType for TEQ',
     queue_payload_type =>'Message_type',
     queue_properties   =>null,
     replication_mode   =>null);
 DBMS_AQADM.START_QUEUE (queue_name=> 'objType_TEQ', enqueue =>TRUE, dequeue=> True);
END;
/
```

なおデータ・タイプは「queue_payload_type」で指定しますが、以下のタイプを選択できます。
+ Message_type
+ RAW
+ JSON
+ JMS

次にサブスクライバーを作成します。
[ADD_SUBSCRIBERプロシージャ](https://docs.oracle.com/cd/F39414_01/arpls/DBMS_AQADM.html#GUID-2B4498B0-7851-4520-89DD-E07FC4C5B2C7)
[AQ$_AGENTタイプ](https://docs.oracle.com/cd/F39414_01/arpls/advanced-queuing-AQ-types.html#GUID-2F1133F1-6F59-4379-8493-923C042E7042)

```sql
BEGIN
DBMS_AQADM.ADD_SUBSCRIBER(
     queue_name => 'objType_TEQ',
     subscriber => sys.aq$_agent('teqBasicObjSubscriber', null, 0),
     rule => 'correlation = ''teqBasicObjSubscriber''');
END;
/
```     

これで事前準備が出来ましたので、メッセージをエンキューします。
今回は時刻情報をメッセージとしてエンキューしてみます。
[ENQUEUEプロシージャ](https://docs.oracle.com/cd/F39414_01/arpls/DBMS_AQ.html#GUID-E262FFC1-2B21-425A-914C-B58238198455)

```sql
DECLARE
 enqueue_options     dbms_aq.enqueue_options_t;
 message_properties  dbms_aq.message_properties_t;
 message_handle      RAW(16);
 message             Message_type;
 dateinfo            VARCHAR(20);

BEGIN
 SELECT TO_CHAR(sysdate,'YYYY-MM-DD HH24:MI:SS') INTO dateinfo FROM dual;
 message := Message_type('TIME', dateinfo);
 message_properties.correlation := 'teqBasicObjSubscriber';

 DBMS_AQ.ENQUEUE(
     queue_name           => 'objType_TEQ',
     enqueue_options      => enqueue_options,
     message_properties   => message_properties,
     payload              => message,
     msgid                => message_handle);
    COMMIT;
END;
/
```

メッセージをデキューします。
[DEQUEUEプロシージャ](https://docs.oracle.com/cd/F39414_01/arpls/DBMS_AQ.html#GUID-D95A89C8-516A-4AF9-8F1F-26B4C6E64420)

```sql
SET SERVEROUTPUT ON

DECLARE
    dequeue_options     dbms_aq.dequeue_options_t;
    message_properties  dbms_aq.message_properties_t;
    message_handle      RAW(16);
    message             Message_type;

BEGIN
    dequeue_options.dequeue_mode  := DBMS_AQ.REMOVE;
    dequeue_options.wait          := DBMS_AQ.NO_WAIT;
    dequeue_options.navigation    := DBMS_AQ.FIRST_MESSAGE;
    dequeue_options.consumer_name := 'teqBasicObjSubscriber';

    DBMS_AQ.DEQUEUE(
        queue_name         => 'objType_TEQ',
        dequeue_options    => dequeue_options,
        message_properties => message_properties,
        payload            => message,
        msgid              => message_handle);

    DBMS_OUTPUT.PUT_LINE('Message: ' || message.subject || ' ... ' || message.text );
    COMMIT;
END;
/
```

上記を実行すると以下のように実行結果が標準出力されます。

```
Message: TIME ... 2022-04-21 09:00:24
```

最後に作成したTEQ関連のオブジェクトを削除して元に戻します。

```sql
-- subscriberの削除
BEGIN
DBMS_AQADM.REMOVE_SUBSCRIBER(
     queue_name => 'objType_TEQ',
     subscriber => sys.aq$_agent('teqBasicObjSubscriber', null ,0));
END;
/

-- キューの停止
Execute DBMS_AQADM.STOP_QUEUE (queue_name => 'objType_TEQ');

-- キューの削除
Execute DBMS_AQADM.DROP_TRANSACTIONAL_EVENT_QUEUE(queue_name =>'objType_TEQ', force=> TRUE);

-- 検証用に作成したオブジェクト・タイプ「Message_type」の削除
DROP TYPE Message_type;
```

## まとめ

TEQの概要と簡単な使用手順をご紹介しました。
Oracle Databaseを既に利用されている場合、別に製品を導入することなくキューイングを導入できます。
またTEQにより分散トランザクションの仕組みを代替でき、アーキテクチャをよりシンプルにできます。
TEQは無償の Oracle Database 21c XE でも利用できますので、気軽にお試しください！