リシーケンス・メッセージ

リシーケンス・メッセージ技術アクセラレータは、システムに入るメッセージが特定の順序で処理されるようにする手段を提供します。

ノート:

Oracle Integrationには、リシーケンス・メッセージ技術アクセラレータが自動的にインストールされています。したがって、このアクセラレータは、サインイン時にアクセラレータおよびレシピ・ページにすでにINSTALLEDとしてリストされます。

内容は次のとおりです。

説明

多くの場合、メッセージが厳密な順序で処理されるようにする必要があります。 Oracle Integrationおよびリシーケンス・メッセージ技術アクセラレータを使用してオーダーを強制するパターンを確認します。このパターンでは、エンドポイント・システムへのコールの同時実行性を制限する必要がある場合も処理されます。

順序付けの問題

基本的な問題は、アカウントの作成、アカウント住所の更新、アカウント担当者の更新など、順番に実行する必要があるリクエストのストリームがあることです。後者の2つのアクティビティは、最初のアクティビティが完了するまで発生しません。

メッセージを順序付けする前に、メッセージの処理順序を把握しておく必要があります。したがって、なんらかの順序IDを使用する必要があります。このIDには、タイムスタンプまたは実際の順序識別子を指定できます。タイムスタンプを使用している場合は、タイムスタンプがより適切に適用されることをメッセージの起点に近づけることができます。たとえば、メッセージがOracle Integrationに到着したときのタイムスタンプを取得した場合、ネットワーク遅延によって、メッセージがすでに順序がずれている可能性があります。

通常、すべてのメッセージを同じ順序で並べる必要はありません。このアカウントの例では、特定のアカウントのメッセージのみをオーダーする必要があります。様々なアカウントのメッセージをパラレルで実行できます。したがって、メッセージ・ストリーム内の様々なシーケンス・ストリームを識別するために、なんらかのグループIDも必要になりました。

メッセージを取得し、その順序を把握したら、それらを処理できます。リシーケンス・ソリューションに固有のものは、メッセージが順不同に到着して順序どおりにソートできるようにするための、ある程度の遅延です。遅延のサイズは、遅延するメッセージを受け入れてから、遅延しないで先に進むことができる時間を指定します。

リシーケンス・ソリューション

リシーケンス・メッセージ・テクニカル・アクセラレータには、標準のOracle Integration機能を使用する一連の統合、接続およびスクリプトが含まれています。メッセージのリシーケンスを処理する統合は汎用です。技術アクセラレータ・パッケージで提供されている統合を変更せずに、技術アクセラレータを使用して様々なタイプのビジネス統合を再順序付けできます。

リシーケンス・メッセージ技術アクセラレータ:

メッセージの到着順序ではなく、目的の順序IDに基づいて入力メッセージを処理します。
一定期間(パーキング時間)ストレージ内の各メッセージをパークして、順序が正しくないメッセージが目的の順序で処理されるようにします。
送信コールを抑制するために、パラレルで処理されるメッセージ・グループの最大数を構成できます。
システム・エラー、ネットワーク・エラー、不正なリクエストなど、すべてのエラー処理を処理します。

前提条件

ソリューションは、データベースを使用して入力メッセージを格納します。 SQLスクリプトを使用して、必要なデータベース表を作成できます。

必要なデータベース表を作成するには:

「このブログ」にあるDDL SQL scriptを検索してダウンロードします。
スクリプトを実行してデータベース表を作成します。

キー・パラメータ

リシーケンス・メッセージ技術アクセラレータでは、次のキー・パラメータを使用してメッセージの順序を変更します。

フィールド	説明
gtype - グループ・タイプ	ストリームのタイプ。様々なメッセージ・タイプを並行して順序付けできます。たとえば、アカウントの更新と担当者の更新は様々なグループ・タイプです。
gid - メッセージ・グループ	順序付けされるメッセージの特定のストリームを識別するリクエスト内のフィールド。
id - メッセージ識別子	このメッセージの一意の識別子。
sequenceId - メッセージ順序	ストリーム内のメッセージの順序付け方法を決定するために使用される、リクエスト内のフィールドまたはタイムスタンプ。
駐車時間	メッセージが目的の順序で処理されるようにするために、メッセージが遅延される可能性がある時間。
メッセージ同時実行性	パラレルで処理されるメッセージ・グループの最大数。

接続

リシーケンス・メッセージ技術アクセラレータでは、次の接続が使用されます。アクセラレータをインストールした後、各接続を構成する必要があります。

接続	タイプ	説明
RSQ DB	Invoke	リシーケンサで使用されるATPデータベース
RSQMessageConsumer	トリガーと呼び出し	メッセージ・コンシューマへのコールのロード分散に使用されます。
RSQManager	トリガー	マネージャ・インタフェースの起動に使用
RSQProducer	トリガー	プロデューサ・インタフェースの起動に使用
RSQDispatcher	トリガー	ディスパッチャ・インタフェースの起動に使用
TestService	Invoke	サンプル・テスト・サービスの起動に使用されます。

アーキテクチャ

リシーケンス・メッセージ技術アクセラレータのアーキテクチャを見てみましょう。

「図reseq-msgs-architecture.pngの説明」

ビジネス・フロントエンド統合

この層の統合は、リシーケンサおよび実際のビジネス統合のフロントエンドとして機能するため、メッセージは順序変更するリシーケンサ統合を通過できます。フロントエンド統合はユースケースに固有であり、型付きリクエストをグループ・フィールドや順序付けフィールドなどの共通フォーマットに変換する手段として機能します。

また、フロントエンド統合は、標準化されたプロデューサ・メッセージを呼び出す前に、メッセージから型付きビジネス・ペイロード、抽出IDおよびグループIDを受け取ります。

リシーケンサ統合

リシーケンス・メッセージ技術アクセラレータには、リシーケンサ統合のセットが含まれています。これらの統合は、メッセージのリシーケンスを処理し、汎用的です。これらを使用して、様々なタイプのビジネス統合を再順序付けできます。追加の変更は必要ありません。

リシーケンサ統合説明

リシーケンサ統合	説明
Producer	リシーケンサのエントリ・ポイントとして機能します。 producer統合はリシーケンス・メッセージを受信し、グループ表に新しい行がない場合は作成し、グループのステータスをNに設定します。次に、メッセージ表にメッセージを作成します。サンプル・メッセージ・ペイロード: `{ "gid": "gid", "gtype": "order", "id": "mid", "sequenceId": 123, "payload": "string repesentation of the payload" }`
グループ・コンシューマ	アクティブなグループを検出し、「メッセージ・コンシューマ」統合を起動します。統合は毎分実行されるようにスケジュールされています。スケジュールを設定する場合は、次の式を使用します: `FREQ=MINUTELY;INTERVAL=1;` グループ・コンシューマ統合では、アクティブなグループが検出され、ターゲット・システムの過負荷を防ぐために送信コールを抑制する並列性が制限されます。アクティブなグループごとに、統合によってメッセージ・コンシューマが起動されます。
メッセージ・コンシューマ	指定されたグループのアクティブなメッセージを処理します。グループ・コンシューマ統合からグループIDおよびタイプを受け取ります。 sequenceIDによって順序付けられたグループのアクティブ・メッセージをロードします。メッセージは、少なくとも駐車時間と同じ経過時間である必要があります。これにより、メッセージが順不同に到着し、順番に処理されるウィンドウが作成されます。統合では、アクティブなメッセージをループし、メッセージ・ステータスをPとしてマークし、ディスパッチャを起動します。ここでは例外が発生する可能性があることに注意してください。ディスパッチャが特定のメッセージに対して戻った後、そのメッセージは削除され、グループ・ステータスが更新されて、アクティブなメッセージがない場合はグループ・ステータスがCにマークされ、新しいアクティブなメッセージがある場合はNにマークされます。サンプル・メッセージ・ペイロード: `{ "id": "Engineering", "type": "employee" }`

Producer

リシーケンサのエントリ・ポイントとして機能します。 producer統合はリシーケンス・メッセージを受信し、グループ表に新しい行がない場合は作成し、グループのステータスをNに設定します。次に、メッセージ表にメッセージを作成します。

サンプル・メッセージ・ペイロード:

{ "gid": "gid", "gtype": "order", "id": "mid",
      "sequenceId": 123, "payload": "string repesentation of the payload"
    }

グループ・コンシューマ

アクティブなグループを検出し、「メッセージ・コンシューマ」統合を起動します。統合は毎分実行されるようにスケジュールされています。スケジュールを設定する場合は、次の式を使用します:

FREQ=MINUTELY;INTERVAL=1;

グループ・コンシューマ統合では、アクティブなグループが検出され、ターゲット・システムの過負荷を防ぐために送信コールを抑制する並列性が制限されます。アクティブなグループごとに、統合によってメッセージ・コンシューマが起動されます。

メッセージ・コンシューマ

指定されたグループのアクティブなメッセージを処理します。グループ・コンシューマ統合からグループIDおよびタイプを受け取ります。 sequenceIDによって順序付けられたグループのアクティブ・メッセージをロードします。メッセージは、少なくとも駐車時間と同じ経過時間である必要があります。これにより、メッセージが順不同に到着し、順番に処理されるウィンドウが作成されます。

統合では、アクティブなメッセージをループし、メッセージ・ステータスをPとしてマークし、ディスパッチャを起動します。ここでは例外が発生する可能性があることに注意してください。ディスパッチャが特定のメッセージに対して戻った後、そのメッセージは削除され、グループ・ステータスが更新されて、アクティブなメッセージがない場合はグループ・ステータスがCにマークされ、新しいアクティブなメッセージがある場合はNにマークされます。

サンプル・メッセージ・ペイロード:

{ "id": "Engineering", "type": "employee"
    }

マネージャ

manager統合によってリシーケンサが監督されます。 3つの操作をサポートしています。

操作パスとメソッド説明

操作	パスとメソッド	説明
構成の取得	パス: /configs メソッド: GET	すべてのタイプの構成を返します。呼出しの例: `$ curl https://my.integration.cloud/ic/api/integration/v1/flows/rest/RSQMANAGER/1.0/configs -v -u username:password`
構成の更新	パス: /configs/{type} メソッド: PUT	指定したタイプの構成を更新します。呼出しの例: `$ curl -X PUT https://my.integration.cloud/ic/api/integration/v1/flows/rest/RSQMANAGER/1.0/configs/employee -v -u username:password -H "Content-Type: application/json" -d@config.json` config.jsonの例: `{ "maxConcurrent": 5, "timeWindow": 11 }`
グループのリカバリ	パス: /types/{type}/groups/{group}/recover メソッド: PUT	メッセージ表のスタック・メッセージを削除し、ステータスをNに設定してグループを再アクティブ化します。呼出しの例: `$ curl -X PUT https://my.integration.cloud/ic/api/integration/v1/flows/rest/RSQMANAGER/1.0/types/employee/groups/eng/recover -v -u username:password -H "Content-length: 0"`

構成の取得

パス: /configs

メソッド: GET

すべてのタイプの構成を返します。呼出しの例:

$ curl https://my.integration.cloud/ic/api/integration/v1/flows/rest/RSQMANAGER/1.0/configs -v -u username:password

構成の更新

パス: /configs/{type}

メソッド: PUT

指定したタイプの構成を更新します。呼出しの例:

$ curl -X PUT https://my.integration.cloud/ic/api/integration/v1/flows/rest/RSQMANAGER/1.0/configs/employee -v -u username:password -H "Content-Type: application/json" -d@config.json

config.jsonの例:

{ "maxConcurrent": 5, "timeWindow": 11 }

グループのリカバリ

パス: /types/{type}/groups/{group}/recover

メソッド: PUT

メッセージ表のスタック・メッセージを削除し、ステータスをNに設定してグループを再アクティブ化します。呼出しの例:

$ curl -X PUT https://my.integration.cloud/ic/api/integration/v1/flows/rest/RSQMANAGER/1.0/types/employee/groups/eng/recover -v -u username:password -H "Content-length: 0"

ディスパッチャ

「ディスパッチャ」は、元のペイロードを再構築して実際のバックエンド統合に送信するリクエスト/レスポンス統合です。リシーケンサ統合とは異なり、ディスパッチャは特定のビジネス統合を呼び出す必要があるため、汎用ではありません。

ディスパッチャはメッセージを受信し、ペイロードを元の型付きビジネス・ペイロードに変換します。グループIDを使用してビジネス・エンドポイントを検索し、同期的に起動します。ここで例外が発生する可能性があります。

ディスパッチャ・インタフェースを次に示します:

{ "id": "Engineering",

"gid" : "Zebra", "gtype" : "order", "sequenceId" : 123,

"payload" : "original payload"

ビジネス統合

「ビジネス統合」は、ビジネス・メッセージを処理する実際の統合です。独自の型付きインタフェースがあります。ビジネス・フロントエンド統合ごとに、対応するビジネス統合が必要です。

エラー処理

ディスパッチャ統合がビジネス統合を呼び出すと、例外が発生する可能性があります。

例外はメッセージ・コンシューマ統合までバブル・アップされ、メッセージ・コンシューマ・インスタンスが失敗します。この場合、グループ・ステータスはデータベース内のPのままになります。

reseq-msgs-error1.pngの説明は以下のとおりです
「図reseq-msgs-error1.pngの説明」

「統合のモニタリング」ページで、失敗したディスパッチャ・インスタンスおよびメッセージ・コンシューマ・インスタンスを確認できます。

「図reseq-msgs-monitor-error2.pngの説明」

システム・エラーのリカバリ

ネットワークの問題などのシステム・エラーが原因で問題が発生した場合は、システム・エラーの解決後に失敗したメッセージ・コンシューマ・インスタンスを再発行することでリカバリできます。

「図reseq-msgs-resubmit-error3.pngの説明」

不正なリクエストのリカバリ

不正なリクエストが原因でエラーが発生した場合、リクエストの再送信は役に立ちません。この場合、不正なリクエストをスキップして続行する必要があります。これを行うには、リシーケンサ・マネージャ統合を起動してスタック・メッセージを削除し、グループを再アクティブ化します:

$ curl -X PUT https://my.integration.cloud/ic/api/integration/v1/flows/rest/RSQMANAGER/1.0/types/employee/groups/eng/recover -v -u username:password -H "Content-length: 0"