Anypoint MQ Consume 操作 - Mule 4

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

受信メッセージ構造

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

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

<anypoint-mq:consume config-ref="Anypoint_MQ_config" destination="myQueue"/>

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

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 メッセージで利用可能なすべての情報をマップします。

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