Oracle Advanced Queuing（ＡＱ）使ってみませんか？ 

Advent Calendar初参加、Qiita初投稿です！


この記事は、[JPOUG Advent Calendar 2019](https://adventar.org/calendars/4154) の16日目の記事です。
15日目はtomoさんの「Flex DiskGroup使ってみたよ！」でした。  
ASMは私も好きな機能ですが、共有ディスクの調達・設定が面倒なので11g以降、実機を使った検証や情報のキャッチアップができていません。自分の勉強不足を改めて認識する記事でした。  


さて先日2019年12月5日に、Oracle Databaseのライセンス緩和により一部オプションがStandard Editionでも使えるようになりました。 

　[Oracle 19cライセンス緩和 機械学習＋地理データがStandard Editionで使用可能に](http://cosol.jp/techdb/2019/12/ml-spatial-is-available-on-oracle-19c-standard-edition.html)

Oracle Databaseだけに限りませんが、いわゆる商用製品（ＯＳ、ミドルウェア）の多くにはエディションの考え方があり、エディションによって（有償または無償で）使える機能・使えない機能があります。 

費用的な理由でエンタープライズ用途でもエントリークラスのエディションのことが多々あるので、上記のようなライセンス緩和は、技術が普及する観点から歓迎されることだと思います。


このエントリーではspatialや機械学習と同様、Oracle Databaseの**全てのエディションで使える機能**の中で、私のお気に入りのひとつ"Advanced Queuing"(AQ)の使い方を紹介します。（業務システムで使っていましたが、やってみた！的な内容です）  


# Oracle Advanced Queuing（AQ）とは
　簡単にいうと非同期でメッセージをやり取りする仕組みです。  
メッセージは同一ノード内で受信・送信することも、（ＤＢリンクを使用して）異なるノード間でも受信・送信することもできます。

　Oracle 9iでAQが導入される以前はPL/SQLパッケージ DBMS_PIPEを使用して非同期にメッセージ送信を行っていました。ただAQと異なりDBMS_PIPEでは同一ノード内（RAC環境）でしか処理ができず、受信側のプロセスを（RACの場合、全ノードで）起動させておく必要がありました。受信側のプロセスを起動し忘れると、もちろん非同期の処理は実行されません。  


　詳細は[マニュアル](https://docs.oracle.com/cd/F19136_01/adque/index.html)をみてください。

# 用途（どんな時に使うのか）
ユーザー・アプリケーションを作成していると、利用者の体感速度を落とさないために、いいかんじで遅延させて実行させたい処理があります。（人間と違って「後でやる」と言ってやらなかったということはコンピュータシステム上許されません）  
例えばＥＣサイトで、発注者や仕入れ先、運送会社などへのFAX送信処理や帳票作成処理などがその例です。（あくまで例です）

　またOracle Databaseの内部ではサーバー生成アラート（キュー名：ALERT_QUE）で使われているそうですが、エラー・ワーニングの発報も有用な用途のひとつだと思われます。


# AQの課題
* 知名度が低い（Oracle Databaseを使っている人でも知らない人が多い印象）
* PL/SQLでしか設定できず導入の敷居が高い（GUIで設定できない）
* 日本語情報が少ない（書籍だと[これ](https://www.amazon.co.jp/dp/4798101702/)くらい？。かなり参考にさせてもらいました）
* 将来のバージョンで非推奨、デサポートされる不安がある。（個人の感想です）  

などがあります。
今回、長年使ってきたAQの使い方をまとめてみましたので、ご参考ください。



# メッセージのシナリオ

　異なるサーバ１とサーバ２で以下のメッセージを送信・受信するとします。

| id | mesg |
|:--:|:--|
| 1 | こんにちは |
| 2 | 今晩は |
| 3 | おはようございます |


　※今回は１インスタンスで試すため同一データベース内でユーザーを分けて受送信します。（USER1とUSER2）  
　18c XE（18.4.0.0.0）で試しました。Always Freeで試そうと思いましたがいろいろ躓くと思ってやめました。


# メッセージ連携する時に事前に決めることは以下の７点です。

* スキーマ　　　　（送信側スキーマ、受信側スキーマ）  
* ＤＢリンク名　　（送信側のみ。パブリックＤＢリンクの必要あり）  
* ユーザー定義型　（送信側と受信側は同じ定義・名前。メッセージの構造を定義）  
* キューテーブル名（送信側と受信側は同じ定義・名前）  
* キュー名　　　　（送信側と受信側は同じ定義・名前）  
* サブスクライバ名（送信側と受信側は同じ定義・名前。送信先の識別子）  
* コールバック関数（受信側のみで定義。非同期で行う業務処理を定義）  

　ユーザー定義型とコールバック関数以外は、任意の名称で問題ありませんが、大文字アルファベット、数字のみで定義することをお勧めします。  
小文字が混ざっているとうまく設定できないことがありました。今回は以下のようにします。  

* スキーマ　　　　　USER1（送信側）、USER2（受信側）  
* ＤＢリンク名　　　DBLINK_USER2  
* ユーザー定義型　　MYMESG  
* キューテーブル名　MYMESG_TAB  
* キュー名　　　　　MYMESG_Q  
* サブスクライバ名　SUBSCR  
* コールバック関数　AUTO_RECV  

　※USER1/USER2は作成済であるとします。


# 設定手順例

## 送信側の設定
###  （送信側）パブリックＤＢリンクを作成
	SQL> CREATE PUBLIC DATABASE LINK DBLINK_USER2 CONNECT TO USER2 IDENTIFIED BY USER2 USING 'ORCL';

### （送信側）権限付与
	SQL> GRANT AQ_ADMINISTRATOR_ROLE TO USER1;

### （送信側）ユーザー定義型、キュー表、キューを作成しキューを開始する
	SQL> CREATE TYPE MYMESG AS OBJECT (
	           ID    NUMBER(10,0),
	           MESG  VARCHAR2(30)
	         )
	/
	SQL> exec DBMS_AQADM.CREATE_QUEUE_TABLE( -
	           queue_table        => 'MYMESG_TAB', -
	           queue_payload_type => 'MYMESG', -
	           multiple_consumers => TRUE -
	       );
	SQL> exec DBMS_AQADM.CREATE_QUEUE( -
	           queue_name  => 'MYMESG_Q', -
	           queue_table => 'MYMESG_TAB' -
	       );
	SQL> exec DBMS_AQADM.START_QUEUE ( -
	           queue_name  => 'MYMESG_Q' -
	       );


## 受信側の設定
### （受信側）権限付与
	SQL> GRANT AQ_ADMINISTRATOR_ROLE TO USER2;
	SQL> GRANT EXECUTE ON DBMS_AQ    TO USER2;

### （受信側）ユーザー定義型、キュー表、キューを作成しキューを開始する（※送信側と同じです。）
  

	SQL> CREATE TYPE MYMESG AS OBJECT (
	           ID    NUMBER(10,0),
	           MESG  VARCHAR2(30)
	         )
	/
	SQL> exec DBMS_AQADM.CREATE_QUEUE_TABLE( -
	                queue_table        => 'MYMESG_TAB', -
	                queue_payload_type => 'MYMESG', -
	                multiple_consumers => TRUE -
	         );
	SQL> exec DBMS_AQADM.CREATE_QUEUE( -
                  queue_name  => 'MYMESG_Q', -
                  queue_table => 'MYMESG_TAB' -
         );
	SQL> exec DBMS_AQADM.START_QUEUE ( -
                  queue_name  => 'MYMESG_Q' -
         );

### （受信側）サブスクライバ（購読者）を登録
	SQL> exec DBMS_AQADM.ADD_SUBSCRIBER( -
	                queue_name  => 'MYMESG_Q', -
	                subscriber  => SYS.AQ$_AGENT('SUBSCR',NULL,NULL), -
	                rule        => NULL -
	         );


## 送信側の設定
### （送信側）サブスクライバ（購読者）を登録
	SQL> exec DBMS_AQADM.ADD_SUBSCRIBER( -
	                queue_name  => 'MYMESG_Q', -
	                subscriber  => SYS.AQ$_AGENT('SUBSCR','USER2.MYMESG_Q@DBLINK_USER2',NULL), -
	                rule        => NULL -
	         );

### （送信側）伝播（リモートインスタンスへの転送スケジュール）の開始
	SQL> exec DBMS_AQADM.SCHEDULE_PROPAGATION( -
	           queue_name    => 'MYMESG_Q', -
	           destination   => 'DBLINK_USER2', -
	           duration      => NULL, -
	           latency       => 0 -
	         );
　　※伝播の開始は、ＤＢリンク先毎に最初の１回実行すればよいです。




## 受信側の設定
### （受信側）受信側でメッセージを受け取った時に、自動でデキューするプロシージャを作成。
	SQL> CREATE OR REPLACE PROCEDURE USER2.AUTO_RECV(
	             context    IN RAW,
	             reginfo    IN SYS.AQ$_REG_INFO,
	             descr      IN SYS.AQ$_DESCRIPTOR,
	             payload    IN VARCHAR2,
	             payloadl   IN NUMBER
	         )IS
	             mesg    MYMESG;
	             dequeue_options     DBMS_AQ.DEQUEUE_OPTIONS_T;
	             message_properties  DBMS_AQ.MESSAGE_PROPERTIES_T;
	             message_handle      RAW(16);
	         BEGIN
	             -- サブスクライバ名を設定
	             dequeue_options.CONSUMER_NAME := 'SUBSCR';
	             DBMS_AQ.DEQUEUE(queue_name         => 'SCOTT.MYMESG_Q',
	                             dequeue_options    => dequeue_options,
	                             message_properties => message_properties,
	                             payload            => mesg,
	                             msgid              => message_handle);
	             --
	             -- ここから業務処理を実装する。今回はINSERTのみする。
	             --
	             INSERT INTO LOG_MESG VALUES(mesg.ID,  mesg.MESG);
	             COMMIT;
	         EXCEPTION
	             WHEN OTHERS THEN Null;
	         END AUTO_RECV;
	/

### （受信側）上記のプロシージャを自動デキューするように登録。
	SQL> DECLARE
	             REGINFO  SYS.AQ$_REG_INFO;
	             REGINFOS SYS.AQ$_REG_INFO_LIST;
         BEGIN
             REGINFO := SYS.AQ$_REG_INFO(
                         'USER2.MYMESG_Q:SUBSCR',
                          DBMS_AQ.NAMESPACE_AQ,
                          'plsql://USER2.AUTO_RECV?PR=1',
                          NULL
                        );
             REGINFOS := SYS.AQ$_REG_INFO_LIST(REGINFO);
             DBMS_AQ.REGISTER( REGINFOS,1);
         END;
	/


# エンキュー（送信）の例
以上で、設定は終了です。
エンキュー（メッセージ送信）は、以下のようなPL/SQLを実行します。  
実際は、この処理をラップするプロシージャ、ファンクションを作ると便利です。  


## コマンド例
	DECLARE
	    mesg MYMESG;
	    enqueue_options     dbms_aq.enqueue_options_t;
	    message_properties  dbms_aq.message_properties_t;
	    message_handle      RAW(16);
	BEGIN
		-- ひとつめのメッセージを送信
	    mesg := MYMESG( 1 ,'こんにちは' );
	    dbms_aq.enqueue(queue_name         => 'MYMESG_Q',
	                    enqueue_options    => enqueue_options,
	                    message_properties => message_properties,
	                    payload            => mesg,
	                    msgid              => message_handle
	    );
		-- ふたつめのメッセージを送信
	    mesg := MYMESG( 2 ,'今晩は' );
	    dbms_aq.enqueue(queue_name         => 'MYMESG_Q',
	                    enqueue_options    => enqueue_options,
	                    message_properties => message_properties,
	                    payload            => mesg,
	                    msgid              => message_handle
	    );
	    -- トランザクションを終了（COMMIT or ROLLBACK)しないとキューに溜められない
	    COMMIT;
	END;
	/

上記を実行すると、自動デキューのプロシージャが実行されLOG_MESG表にメッセージが登録されます。



# まとめ
　本格的な非同期処理を行いたい場合は、専用のミドルウェアを導入すると思いますが、ちょっとしたシステムの場合、Oracle Databaseの標準機能ＡＱを使って実装することができます。    
9iからの機能でありDatabase内部でも使われているため枯れた機能に入ると思います。  

Oracle DatabaseにはAQ以外にもRDBMSの範囲を超えた様々な機能があります。有償オプションの機能は最大限活用されていると思いますが、それ以外の無償機能も使ってみてはいかがでしょうか？。開発生産性が向上すると思います。



**（注意）** 昔（9i）の知識で動作確認した程度なので、実際に使用する場合は、最新のマニュアルを確認して使ってください。  

  
明日は[実行計画は、SQL文のレントゲン写真だ！　Oracle Database編 (全部俺） Advent Calendar 2019](https://adventar.org/calendars/4737)を書かれているHiroshi Sekiguchiさんです。


# おまけ
## 自動デキュー設定を確認する
	SQL> SELECT * FROM SYS.REG$;  



## 自動デキューを解除する
	DECLARE
	    REGINFO  SYS.AQ$_REG_INFO;
	    REGINFOS SYS.AQ$_REG_INFO_LIST;
	BEGIN
	    REGINFO := SYS.AQ$_REG_INFO(
	                 'USER2.MYMESG_Q:SUBSCR',
	                 DBMS_AQ.NAMESPACE_AQ,
	                 'plsql://USER2.AUTO_RECV?PR=1',
	                 ''
	               );
	    REGINFOS := SYS.AQ$_REG_INFO_LIST(REGINFO);
	    DBMS_AQ.UNREGISTER( REGINFOS,1);
	END;
	/

## 手動デキュー
	DECLARE
	    mesg MYMESG;
	    dequeue_options     dbms_aq.dequeue_options_t;
	    message_properties  dbms_aq.message_properties_t;
	    message_handle      RAW(16);
	BEGIN
	    dequeue_options.CONSUMER_NAME := 'SUBSCR';
	
	    DBMS_AQ.DEQUEUE(queue_name         => 'MYMESG_Q',
	                    dequeue_options    => dequeue_options,
	                    message_properties => message_properties,
	                    payload            => mesg,
	                    msgid              => message_handle);
	
	    DBMS_OUTPUT.PUT_LINE ('Message[' || mesg.id || '][' || mesg.mesg || ']');
	    COMMIT;
	END;
	/

## AQをPostgreSQLに移行する場合
OracleからPostgreSQLに移行する場合、PostgreSQLのdblinkとNOTIFY/LISTEN機能を組み合わせることでAQと同様のことができそうです。まだ試していませんが、AQよりDBMS_PIPEに似た機能になります。