1. ホームページ
  2. Anypoint MQ Connector

  3. Anypoint MQ Consume 操作

All Connectors

Anypoint MQ Consume 操作 - Mule 4

Anypoint MQ Connector の Consume 操作を使用すると、アプリケーションはユースケースをサポートする肯定応答戦略を使用して、フローの実行中にいつでもキューから単一メッセージをコンシュームできます。

Anypoint MQ では、メッセージエクスチェンジではなく、キューからのコンシュームがサポートされます。

アプリケーションがメッセージエクスチェンジからコンシュームしようとすると、Anypoint MQ REST API は次のエラーをスローします。 ​The destination should be a queue. (宛先はキューである必要があります。)

受信メッセージ構造

すべての受信メッセージは 2 つの部分から構成されます。

  • payload​ - メッセージ本文を表します。

  • attributes​ - 受信したメッセージのメタデータ (ID や他のプロパティなど) が含まれます。

ペイロードとメッセージ属性を含む Mule メッセージ

メッセージのコンシューム

キューからメッセージをコンシュームする最も簡単な方法は、Mule パレットからキャンバスに ​[Consume]​ 操作をドラッグして、メッセージの取得元であるキューの名前で ​Queue​ パラメーターを設定することです。

宛先、肯定応答モードとタイムアウト、ポーリングタイムアウトを含むコネクタ設定 
<anypoint-mq:consume config-ref="Anypoint_MQ_config" destination="myQueue"/>
xml

この例の Consume 操作は、​myQueue​ で最初に使用可能なメッセージをコンシュームし、次の構造の ​MuleMessage​ インスタンスに変換します。

  • ペイロードとしてのメッセージ本文。

  • メッセージ属性に含まれるメッセージのメタデータ。属性には ​messageId​ または ​redeliveryCount​ が含まれます。

Queue​ パラメーターで DataWeave 式を使用して、メッセージのコンシューム元となる対象キューを動的に変更します。

コンシュームされるメッセージ肯定応答

  • 自動

    デフォルトでは、​acknowledgementMode​ は ​IMMEDIATE​ です。つまり、メッセージはキューからコンシュームされるとすぐに、Mule フローに配信される前に肯定応答されます。メッセージの肯定応答に失敗すると、Consume 操作は ​ANYPOINT-MQ:ACKING​ エラーで終了します。

    詳細は、​「即時肯定応答」​を参照してください。

  • 手動

    メッセージの肯定応答を手動で制御するには、​acknowledgementMode​ を ​MANUAL​ に変更します。​MANUAL​ 肯定応答モードを使用している場合、アプリケーションロジックは ​『ACK』​ または ​『NACK』​ 操作を使用して、メッセージの肯定応答を実行するタイミングを決定します。

    手動肯定応答を実行するには、結果のメッセージ属性の一部として提供される ​ackToken​ の値が必要です。

    詳細は、​「手動肯定応答」​を参照してください。

肯定応答タイムアウトについての詳細は、​「肯定応答タイムアウト」​を参照してください。

空のキュー管理

キューにメッセージがある場合、Consume 操作はすぐにメッセージの 1 つを返します。

キューが空の場合、この操作の動作は ​pollingTime​ パラメーターの値によって異なります。

  • 値なし

    デフォルトでは、Consume 操作は空のキューからコンシュームするときに新しいメッセージの到着を 10000 ミリ秒 (10 秒) 間待機します。

    この時間内に Anypoint MQ でメッセージが受信されない場合、Consume は ​ANYPOINT-MQ:TIMEOUT​ エラーで失敗します。

  • 0

    pollingTime​ の値が ​0​ の場合、Consume 操作はデフォルトの 10000 ミリ秒 (10 秒) 間待機します。

    この時間内に Anypoint MQ でメッセージが受信されない場合、Consume で結果として ​null​ メッセージが生成され、エラーは生成されません。

  • 1​ から ​20000

    pollingTime​ の値が 1 ～ 20000 ミリ秒 (最大値) の場合、Consume は新しいメッセージの到着を、​pollingTime​ で指定された時間待機します。

    特定の時間枠内で Anypoint MQ でメッセージが受信されない場合、Consume は ​ANYPOINT-MQ:TIMEOUT​ エラーで失敗します。

  • pollingTime​ の値が 0 ～ 20000 ミリ秒 (20 秒) の範囲外の場合、Consume は ​ANYPOINT-MQ:CONSUMING​ エラーで失敗します。

出力メッセージのメタデータ

受信する各メッセージは、​Payload​ と ​Attributes​ の 2 つの部分から構成されます。​Attributes​ にはメッセージのメタデータが含まれます。

このメタデータは、次を含む Anypoint MQ メッセージで利用可能なすべての情報をマップします。

  • メッセージ ID

  • ヘッダーマップ

  • プロパティマップ

ペイロードとメッセージ属性を含む Mule メッセージ

attributes​ 構造についての詳細は、​「Anypoint MQ Connector リファレンス - 4.x」​の Consume 操作を参照してください。

関連情報