RabbitMQ メッセージング API の使用法

RabbitMQ との接続

RabbitMQ に接続するには以下の手順に従います。

1. 設定オブジェクトを作成します。そのためには、%External.Messaging.RabbitMQSettingsOpens in a new tab のインスタンスを作成し、必要に応じてそのプロパティを以下のように設定します。

   • username と password ：クライアントの RabbitMQ 認証情報を定義します。これらのプロパティに値を設定しないと、既定で "guest" が指定されます。

   • host ：ホスト サーバの名前を定義します。定義しない場合は、既定値の "localhost" になります。

   • port ：ホスト サーバ側で使用するポート番号を定義する整数です。既定値は 5672 です。

   • virtualHost ：必要に応じて仮想ホストの名前を定義します。

   例えば以下のようにします。

    Set settings = ##class(%External.Messaging.RabbitMQSettings).%New()
    Set settings.username = "ronaldkellogg"
    Set settings.password = "449!ds%t"
    Set settings.host = "bazco.com"
    Set settings.port = 5693
   

   

2. (オプション)SSL/TLS を使用して RabbitMQ に接続する場合は、設定の enableSSL プロパティを 1 に設定したうえで、使用している TLS 構成に応じて以下の各プロパティを設定します。

   • tlsVersion ：使用している TLS プロトコルのバージョンを指定した文字列。既定値は "TLSv1.2" です。

   • enableHostnameVerification ：サーバ証明書に記載されたホスト名とサーバのホスト名が一致することを相手検証プロセスで検証するかどうかを指定するブーリアン値。

   • clientKeyFile ：相手検証を実行するようにサーバを構成している場合に、クライアントの秘密鍵ファイルへのパスを指定する文字列。

   • keyPassword ：クライアントの鍵ファイルがセキュリティで保護されている場合に、そのファイルへのアクセスで必要なパスワード文字列。

   • keyStoreFile ：鍵ストア・ファイルへのパスを指定する文字列。

   • keyStorePassword ：クライアントの鍵ストア・ファイルがセキュリティで保護されている場合に、そのファイルへのアクセスで必要なパスワード文字列。

3. メッセージング・クライアント・オブジェクトを作成します。そのためには、汎用 %External.Messaging.ClientOpens in a new tab クラスの CreateClient() メソッドを呼び出し、設定オブジェクトを 1 番目の引数として渡します。例えば以下のようにします。

    Set client = ##class(%External.Messaging.Client).CreateClient(settings, .tSC)
    If $$$ISERR(tSC) { //handle error scenario }
   

   

   このメソッドからは、参照渡しでステータス・コードが 2 番目の引数として返されます。コードでは、このステータスを確認したうえで処理を継続する必要があります。

   settings オブジェクトは %External.Messaging.RabbitMQSettingsOpens in a new tab のインスタンスなので、返されるオブジェクト (client) は %External.Messaging.RabbitMQClientOpens in a new tab のインスタンスです。

RabbitMQ パブリッシャ

InterSystems IRIS は、メッセージを作成する API メソッドを呼び出し、作成したメッセージを送信することにより、RabbitMQ パブリッシャとして機能できます。メッセージの送信先とするエクスチェンジ、またはエクスチェンジからのメッセージのルーティング先とするキューをアプリケーションで作成する場合は、"エクスチェンジとキューの操作" を参照してください。以下のフローでは、クライアント・オブジェクトを使用し、パブリッシャとして RabbitMQ を操作しています。

メッセージの作成

送信するメッセージを作成するには、%External.Messaging.RabbitMQMessageOpens in a new tab オブジェクトの新しいインスタンスを作成します。つづいて、そのメッセージ・オブジェクトのプロパティを必要に応じて定義します。使用できるメッセージすべての説明は、RabbitMQ のドキュメントOpens in a new tabを参照してください。contentEncoding の値が UTF-8 の場合は SetEncodedContent() メソッドを呼び出し、それ以外の値の場合は SetContent() メソッドを呼び出すことで、メッセージのコンテンツを設定します。例えば以下のようにします。

set exchange = "events_handler"
set routingKey = "quick_start"
set deliveryMode = 2
set body = "MyMessage"

set msg = ##class(%External.Messaging.RabbitMQMessage).%New()
set msg.exchange = exchange
set msg.routingKey = routingKey
set msg.deliveryMode = deliveryMode
do msg.SetEncodedContent(body)

メッセージの送信

作成したメッセージは、RabbitMQ クライアント・オブジェクトの SendMessage() メソッドを実行することにより、トピックに送信できます。例えば以下のようにします。

set tSC = client.SendMessage(msg)
if $$$ISERR(tSC) { //handle error scenario }

RabbitMQ コンシューマ

InterSystems IRIS は、キューからメッセージを取得する API を呼び出すことで、RabbitMQ コンシューマとして機能できます。以下のフローでは、クライアント・オブジェクトを使用し、コンシューマとして RabbitMQ を操作しています。

メッセージ取得のための設定の構成 (オプション)

RabbitMQ クライアントは、ReceiveMessage() メソッドを使用して RabbitMQ コンシューマとして機能できます。このメソッドでは、JSON 形式の文字列をオプションの引数として指定することで、メッセージ取得操作の設定を指定できます。そのためには、%External.Messaging.RabbitMQReceiveSettingsOpens in a new tab クラスの新しいインスタンスを作成し、必要に応じてそのプロパティを設定します。利用できるプロパティは以下のとおりです。

• autoAck ：ブーリアン値。真の場合、サーバでは、メッセージが配信されたときに、そのメッセージが確認応答済みであると自動的に見なされます。偽の場合、サーバでは、コンシューマからの明示的な手動の確認応答があることが想定されます。このプロパティ値を設定していない場合、autoAck には既定値である false が設定されます。

• ackMultiple ：ブーリアン値。真の場合、すべてのメッセージのバッチについて、手動による確認応答が実行されます。この確認応答で指定した配送タグに相当するメッセージも、このバッチの対象となります。偽の場合、確認応答で指定した配送タグに相当するメッセージのみについて、確認応答が実行されます。このプロパティ値を設定していない場合、ackMultiple には既定値である false が設定されます。

例えば以下のようにします。

Set rset = ##class(%External.Messaging.RabbitMQReceiveSettings).%New()
Set rset.autoAck = 0

メッセージの取得

メッセージを取得するには、RabbitMQ クライアントが継承した ReceiveMessage() メソッドを呼び出します。このメソッドは、キューの名前を引数として取り、参照渡しで %ListOfObjectsOpens in a new tab としてメッセージを返します。前のセクションで説明したメッセージ取得設定を指定している場合は、ToJSON() メソッドを使用して、その設定を 3 番目の引数として指定します。例えば以下のようにします。

 #dim messages As %ListOfObjects
 Set tSC = client.ReceiveMessage(queue, .messages, rset.ToJSON())

 For i=1:1:messages.Size {
        Set msg = messages.GetAt(i)
        Write "Message: ", msg.ToJSON(), !
 }

エクスチェンジとキューの操作

InterSystems IRIS には、RabbitMQ のエクスチェンジとキューを管理するための API が用意されています。これには以下が含まれます。

• エクスチェンジの作成または削除

• キューの作成または削除

• エクスチェンジへのキューの結合

ここでは、この API を使用して上記のタスクを実行する方法について説明します。エクスチェンジとキューの詳しい説明は、RabbitMQ のドキュメントで AMQP 0–9–0 プロトコルの概要Opens in a new tabを参照してください。

エクスチェンジの作成または削除

AMQP 0–9–0 の仕様によれば、すべての RabbitMQ メッセージはエクスチェンジに送信する必要があり、そこから本来の送信先キュー (複数可) へルーティングされます。

エクスチェンジの作成

RabbitMQ クライアント・オブジェクトは、エクスチェンジを作成するための CreateExchange() メソッドを備えています。CreateExchange() は、以下の引数を、ここに記した順序で受け入れます。

1. exchangeName ：エクスチェンジに割り当てる名前。

2. exchangeType ：AMQP 0–9–1 エクスチェンジの 4 つの種類である "direct"、"fanout"、"topic"、"headers". のいずれかを指定した文字列。

3. durable ：ブーリアン値。真であれば、サーバが再起動してもエクスチェンジは存続します。偽であれば、サーバの再起動後はエクスチェンジを再作成する必要があります。

4. autoDelete ：ブーリアン値。真であれば、すべてのキューの結合を解除されたエクスチェンジは削除されます。偽であれば、そのようなエクスチェンジは存続します。

例えば以下のようにします。

Set exchangeName = "broadcast"
Set exchangeType = "fanout"
Set durable = 1
Set autoDelete = 0

Set tSC = client.CreateExchange(exchangeName, exchangeType, durable, autoDelete)

エクスチェンジの削除

アプリケーションで RabbitMQ エクスチェンジの名前を引数に指定して RabbitMQ クライアント・オブジェクトの DeleteExchange() メソッドを呼び出すと、そのエクスチェンジを削除できます。

Set tSC = client.DeleteExchange(exchangeName)

キューの作成または削除

RabbitMQ コンシューマがサブスクライブしているキューへエクスチェンジによってメッセージがルーティングされると、RabbitMQ コンシューマはそのメッセージを受領します。

キューの作成

キューを作成するには、RabbitMQ クライアント・オブジェクトの CreateQueue() メソッドを呼び出します。CreateQueue() は、以下の引数を、ここに記した順序で受け入れます。

1. queueName ：キューに割り当てる名前。

2. durable ：ブーリアン値。真であれば、サーバが再起動してもキューは存続します。偽であれば、サーバの再起動後はキューを再作成する必要があります。

3. exclusive ：ブーリアン値。真であれば、キューは 1 つの接続でのみ使用され、その接続が閉じると削除されます。

4. autoDelete ：ブーリアン値。真であれば、すべてのコンシューマがサブスクライブを終了したキューが削除されます。偽であれば、そのようなキューは存続します。

例えば以下のようにします。

 Set queue = "quick-start-events"
 Set durable = 1
 Set exclusive = 0
 Set autoDelete = 0
 Set tSC = client.CreateQueue(queue, durable, exclusive, autoDelete)

また、すべてのメッセージング・プラットフォームに共通のメソッドを使用して、キューを作成することもできます。この代替手法を使用する場合は、%External.Messaging.RabbitMQQueueSettingsOpens in a new tab クラスのインスタンスにキューの設定を定義し、ToJSON() メソッドを使用して、その設定を JSON オブジェクトとして共通メソッドに渡します。詳細は、"%External.Messaging.Client.CreateQueueOrTopic()Opens in a new tab" を参照してください。

キューの削除

アプリケーションで RabbitMQ キューの名前を引数に指定して RabbitMQ クライアント・オブジェクトの DeleteQueue() メソッドを呼び出すと、その RabbitMQ キューを削除できます。

Set tSC = client.DeleteQueue(queueName)

また、すべてのメッセージング・プラットフォームに共通のメソッドを使用して、キューを削除することもできます。詳細は、"%External.Messaging.Client.DeleteQueueOrTopic()Opens in a new tab" を参照してください。

エクスチェンジへのキューの結合

キューにメッセージをルーティングするには、キューをエクスチェンジに結合する必要があります。アプリケーションで RabbitMQ クライアント・オブジェクトの BindQueue() メソッドを呼び出すことで、エクスチェンジにキューを結合できます。

BindQueue() メソッドでは、キュー名、エクスチェンジ名、および複数の結合キーをコンマ区切りで記述した文字列を引数として受け入れます。結合キーの中で使用しているコンマは、その前に円記号 (\) を記述することでエスケープできます。結合キーの中で使用している円記号は、その前にもう 1 つ円記号を記述してエスケープできます。

例えば、必要な 2 つの結合キーとして "event-log,critical" と "event-log,urgent/important" があるとすると、アプリケーションのコードでは次のようにしてキューを結合できます。

Set bindingKeys = "event-log\,critical,event-log\,urgent\\important"
Set tSC = client.BindQueue(queueName, exchangeName, bindingKeys)

クライアントの終了

RabbitMQ との通信を完了した InterSystems IRIS アプリケーションは、Close() メソッドを使用してクライアントを閉じる必要があります。例えば以下のようにします。

 Do:client'="" client.Close()