# mulesoft Anypoint Mq Connector - Full Documentation # Anypoint MQ ACK および NACK 操作 メッセージの肯定応答 (ACK) は、メッセージが処理されたので再配信されないようにキューから削除する必要があることを Anypoint MQ ブローカーに通知します。メッセージの否定応答 (NACK) は、メッセージを処理せずに、任意のコンシューマーに再配信するためにキューに戻すようブローカーに通知します。 ## [](#肯定応答モード)肯定応答モード Anypoint MQ Connector には 3 つの肯定応答モードがあります。 - Immediate (即時) アプリケーションがメッセージを処理する前にコンシュームするときにそのメッセージの肯定応答を行うには、即時モードを使用します。 - 自動 フローの実行が正常に完了した場合にのみ受信メッセージの自動的な肯定応答を行うには、自動モードを使用します。 - 手動 メッセージの肯定応答を行うタイミングを決定するアプリケーションロジックにすべての責任を委譲するには、手動モードを使用します。 ### [](#即時肯定応答)即時肯定応答 `IMMEDIATE`​ 肯定応答モードを使用して、アプリケーションがメッセージを処理する前にコンシュームするときにそのメッセージの肯定応答を行います。 Anypoint MQ はメッセージの肯定応答を行うと、キューからそのメッセージを削除し、処理中にエラーが発生した場合は再配信しません。 Unresolved include directive in modules/ROOT/pages/anypoint-mq-ack.adoc - include::connectors::partial$mq-disclaimer.adoc\[\] `IMMEDIATE`​ モードを使用したメッセージの処理中にエラーが発生した場合、Anypoint MQ によって ​`ANYPOINT-MQ:ACKING`​ エラーがスローされます。 メッセージを失うことなくエラーを管理するには、デッドレターキューを使用します。 ### [](#自動肯定応答)自動肯定応答 `AUTO`​ 肯定応答モードを使用して、フローの実行が正常に完了した場合にのみ受信メッセージの自動的な肯定応答を行います。 フローの実行中にエラーが発生すると、メッセージは否定応答 (NACK) され、再配信のためにキューに戻されます。 ### [](#手動肯定応答)手動肯定応答 `MANUAL`​ 肯定応答モードを使用して、アプリケーションロジックにメッセージの肯定応答のすべての責任を委譲します。 この設定を使用すると、受信したすべてのメッセージでは ​`ackToken`​ 値が含まれています。メッセージはそれによって特定の接続で一意に識別されます。 この ​`ackToken`​ 値を保存して、後で ACK または NACK 操作への入力として使用します。 ## [](#肯定応答タイムアウト)肯定応答タイムアウト `AUTO`​ または ​`MANUAL`​ 肯定応答モードを使用する場合、​`acknowledgementTimeout`​ パラメーターを使用して、自動的にキューに戻る前にメッセージが肯定応答を待機する時間を制御します。 `acknowledgementTimeout`​ タイマーはメッセージがコンシュームされたときに開始されます。 `acknowledgementTimeout`​ が設定されていない場合、Anypoint MQ Connector では 2 分が使用されます。 別の TTL が必要な場合は、サブスクライバーの ​`acknowledgementTimeout`​ パラメーターでそれを明示的に定義する必要があります。 `acknowledgementTimeout`​ を使用する場合、外部システムの遅延や高負荷によるアプリケーションのバックプレッシャーなどの予期しない遅延の時間を含む、予想されるアプリケーション時間処理を考慮してください。たとえば、コンシュームされたメッセージが 10 秒で処理されると予想される場合は、​`acknowledgementTimeout`​ を 15 秒以上に設定します。

Subscriber ソースで ​prefetch​ モードを使用する場合、Anypoint MQ は、​acknowledgementTimeout​ で指定された有効期限 (時間) を順守しようとします。 ただし、メッセージがバッファに保持されている時間がタイムアウト値の 80% に達すると、Anypoint MQ は有効期限タイムアウトを延長します。

このため、​acknowledgementTimeout​ を順守する必要がある場合は、​prefetch​ モードを使用しないでください。

### [](#肯定応答タイムアウト-auto-肯定応答モード)肯定応答タイムアウト — AUTO 肯定応答モード サブスクライバーがメッセージをコンシュームした後にタイマーが起動し、肯定応答を待機中のブローカーでメッセージがインフライトである期間を示します。​`acknowledgementTimeout`​ 制限に達する前にフロー処理が成功または失敗で終了しない場合、メッセージは再配信のために自動的にキューに戻されます。 ### [](#肯定応答タイムアウト-manual-肯定応答モード)肯定応答タイムアウト — MANUAL 肯定応答モード Subscriber または Consume 操作で ​`MANUAL`​ 肯定応答モードを使用してメッセージをコンシュームしている場合、アプリケーションは ​`acknowledgementTimeout`​ 制限に達する前にメッセージ ​`ackToken`​ を使用して、ACK または NACK 操作を実行する必要があります。タイムアウトの前に指定したトークンでどちらの操作も実行されないと、メッセージは再配信のために自動的にキューに戻されます。タイムアウト制限後にメッセージの肯定応答を試行すると、期限切れのトークンが原因で失敗します。 ## [](#ack-操作-manual-肯定応答モード)ACK 操作 — MANUAL 肯定応答モード メッセージの肯定応答 (ACK) は、メッセージが処理されたので再配信されないようにキューから削除する必要があることをブローカーに通知します。 手動モードを使用して ACK 操作を実行する場合、肯定応答トークン (​`ackToken`​) を使用してメッセージを識別する必要があります。これはメッセージ属性の ​`ackToken`​ 項目にあります。 ```xml ``` メッセージの処理中に void でない操作が起動された場合、Mule メッセージのペイロードと属性が変更されます。処理後に ACK 操作を実行するには、変数に ​`ackToken`​ を保存する必要があります。 後で使用するために属性を保存するには、​`target`​ および ​`targetValue`​ パラメーターを使用して変数にメッセージ全体を保存します。 ```xml #[vars.mqMessage.payload] ``` ## [](#nack-操作)NACK 操作 メッセージの否定応答 (NACK) は、メッセージが正常に処理されなかったので、任意のコンシューマーに再配信するためにキューに戻すようブローカーに通知します。 肯定応答トークン ​`ackToken`​ は、NACK 操作を実行するときに使用する必要があるメッセージの一意の識別子です。これはメッセージ属性の ​`ackToken`​ 項目にあります。 ```xml ``` ## [](#関連情報)関連情報 - [Anypoint MQ Consume 操作](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-consume) - [Anypoint MQ Subscriber ソース](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-listener) # Anypoint MQ ACK および NACK 操作 メッセージの肯定応答 (ACK) は、メッセージが処理されたので再配信されないようにキューから削除する必要があることを Anypoint MQ ブローカーに通知します。メッセージの否定応答 (NACK) は、メッセージを処理せずに、任意のコンシューマーに再配信するためにキューに戻すようブローカーに通知します。 ## [](#肯定応答モード)肯定応答モード Anypoint MQ Connector には 3 つの肯定応答モードがあります。 - Immediate (即時) アプリケーションがメッセージを処理する前にコンシュームするときにそのメッセージの肯定応答を行うには、即時モードを使用します。 - 自動 フローの実行が正常に完了した場合にのみ受信メッセージの自動的な肯定応答を行うには、自動モードを使用します。 - 手動 メッセージの肯定応答を行うタイミングを決定するアプリケーションロジックにすべての責任を委譲するには、手動モードを使用します。 ### [](#即時肯定応答)即時肯定応答 `IMMEDIATE`​ 肯定応答モードを使用して、アプリケーションがメッセージを処理する前にコンシュームするときにそのメッセージの肯定応答を行います。 Anypoint MQ はメッセージの肯定応答を行うと、キューからそのメッセージを削除し、処理中にエラーが発生した場合は再配信しません。 Unresolved include directive in modules/ROOT/pages/anypoint-mq-ack.adoc - include::connectors::partial$mq-disclaimer.adoc\[\] `IMMEDIATE`​ モードを使用したメッセージの処理中にエラーが発生した場合、Anypoint MQ によって ​`ANYPOINT-MQ:ACKING`​ エラーがスローされます。 メッセージを失うことなくエラーを管理するには、デッドレターキューを使用します。 ### [](#自動肯定応答)自動肯定応答 `AUTO`​ 肯定応答モードを使用して、フローの実行が正常に完了した場合にのみ受信メッセージの自動的な肯定応答を行います。 フローの実行中にエラーが発生すると、メッセージは否定応答 (NACK) され、再配信のためにキューに戻されます。 ### [](#手動肯定応答)手動肯定応答 `MANUAL`​ 肯定応答モードを使用して、アプリケーションロジックにメッセージの肯定応答のすべての責任を委譲します。 この設定を使用すると、受信したすべてのメッセージでは ​`ackToken`​ 値が含まれています。メッセージはそれによって特定の接続で一意に識別されます。 この ​`ackToken`​ 値を保存して、後で ACK または NACK 操作への入力として使用します。 ## [](#肯定応答タイムアウト)肯定応答タイムアウト `AUTO`​ または ​`MANUAL`​ 肯定応答モードを使用する場合、​`acknowledgementTimeout`​ パラメーターを使用して、自動的にキューに戻る前にメッセージが肯定応答を待機する時間を制御します。 `acknowledgementTimeout`​ タイマーはメッセージがコンシュームされたときに開始されます。 `acknowledgementTimeout`​ が設定されていない場合、Anypoint MQ Connector では 2 分が使用されます。 別の TTL が必要な場合は、サブスクライバーの ​`acknowledgementTimeout`​ パラメーターでそれを明示的に定義する必要があります。 `acknowledgementTimeout`​ を使用する場合、外部システムの遅延や高負荷によるアプリケーションのバックプレッシャーなどの予期しない遅延の時間を含む、予想されるアプリケーション時間処理を考慮してください。たとえば、コンシュームされたメッセージが 10 秒で処理されると予想される場合は、​`acknowledgementTimeout`​ を 15 秒以上に設定します。

Subscriber ソースで ​prefetch​ モードを使用する場合、Anypoint MQ は、​acknowledgementTimeout​ で指定された有効期限 (時間) を順守しようとします。 ただし、メッセージがバッファに保持されている時間がタイムアウト値の 80% に達すると、Anypoint MQ は有効期限タイムアウトを延長します。

このため、​acknowledgementTimeout​ を順守する必要がある場合は、​prefetch​ モードを使用しないでください。

### [](#肯定応答タイムアウト-auto-肯定応答モード)肯定応答タイムアウト — AUTO 肯定応答モード サブスクライバーがメッセージをコンシュームした後にタイマーが起動し、肯定応答を待機中のブローカーでメッセージがインフライトである期間を示します。​`acknowledgementTimeout`​ 制限に達する前にフロー処理が成功または失敗で終了しない場合、メッセージは再配信のために自動的にキューに戻されます。 ### [](#肯定応答タイムアウト-manual-肯定応答モード)肯定応答タイムアウト — MANUAL 肯定応答モード Subscriber または Consume 操作で ​`MANUAL`​ 肯定応答モードを使用してメッセージをコンシュームしている場合、アプリケーションは ​`acknowledgementTimeout`​ 制限に達する前にメッセージ ​`ackToken`​ を使用して、ACK または NACK 操作を実行する必要があります。タイムアウトの前に指定したトークンでどちらの操作も実行されないと、メッセージは再配信のために自動的にキューに戻されます。タイムアウト制限後にメッセージの肯定応答を試行すると、期限切れのトークンが原因で失敗します。 ## [](#ack-操作-manual-肯定応答モード)ACK 操作 — MANUAL 肯定応答モード メッセージの肯定応答 (ACK) は、メッセージが処理されたので再配信されないようにキューから削除する必要があることをブローカーに通知します。 手動モードを使用して ACK 操作を実行する場合、肯定応答トークン (​`ackToken`​) を使用してメッセージを識別する必要があります。これはメッセージ属性の ​`ackToken`​ 項目にあります。 ```xml ``` メッセージの処理中に void でない操作が起動された場合、Mule メッセージのペイロードと属性が変更されます。処理後に ACK 操作を実行するには、変数に ​`ackToken`​ を保存する必要があります。 後で使用するために属性を保存するには、​`target`​ および ​`targetValue`​ パラメーターを使用して変数にメッセージ全体を保存します。 ```xml #[vars.mqMessage.payload] ``` ## [](#nack-操作)NACK 操作 メッセージの否定応答 (NACK) は、メッセージが正常に処理されなかったので、任意のコンシューマーに再配信するためにキューに戻すようブローカーに通知します。 肯定応答トークン ​`ackToken`​ は、NACK 操作を実行するときに使用する必要があるメッセージの一意の識別子です。これはメッセージ属性の ​`ackToken`​ 項目にあります。 ```xml ``` ## [](#関連情報)関連情報 - [Anypoint MQ Consume 操作](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-consume) - [Anypoint MQ Subscriber ソース](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-listener) # Anypoint MQ Connector 3.x — Mule 4 サポートカテゴリ: Select Anypoint MQ Connector は、Anypoint MQ クラウドサービスを使用して Mule アプリケーションにパブリッシュ - サブスクライブメッセージングを提供します。 Anypoint MQ は、エンタープライズメッセージングパターン、マイクロサービス間の非同期通信、IoT などの広範なメッセージングのユースケースをサポートします。また、Anypoint MQ には、node.js、Java、Go、DevOps bash スクリプトなどのフレームワークを使用してメッセージのパブリッシュやコンシュームを行う REST API もあります。 ## [](#要件)要件 このコネクタを使用するには、Anypoint Studio、Anypoint Connector、アプリケーションフローに精通している必要があります。Anypoint MQ Connector は無料ですが、Anypoint Platform でサービスに接続するにはライセンスが必要です。 ### [](#互換性)互換性
ソフトウェア バージョン

Mule Runtime Engine

4.1.1 以降

Anypoint Studio

7 以降

このドキュメントは、Anypoint MQ Connector 3.x のみに適用されます。バージョン 2.x からバージョン 3.x に移行するには、[Anypoint MQ Connector 3.x 移行ガイド](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-new-3x-major)を参照してください。 ## [](#anypoint-mq-connector-をインストールする)Anypoint MQ Connector をインストールする このコネクタは、Anypoint Studio で記述されたアプリケーション、スタンドアロンアプリケーション、または XML で記述されたアプリケーションで使用できます。 Anypoint Studio で、次の手順を実行します。 1. Mule プロジェクトを作成するか開きます。 2. Studio タスクバーの ​**Exchange**​ アイコンをクリックします。 3. Anypoint Platform のユーザー名とパスワードを入力し、​**\[Sign In (サインイン)\]**​ をクリックします。 4. **\[All assets (すべてのアセット)\]**​ を選択して「MQ」を検索します。 5. \[Anypoint MQ Connector — Mule 4\] を選択し、​**\[Add to project (プロジェクトに追加)\]**​ をクリックします。 6. 画面の指示に従ってコネクタをインストールします。 スタンドアロンアプリケーションまたは XML で記述されたアプリケーションの場合。 1. Anypoint Studio で Mule プロジェクトを開きます。 2. コネクタを連動関係として pom.xml ファイルに追加します。 ```xml com.mulesoft.connectors anypoint-mq-connector x.x.x mule-plugin ``` `x.x.x`​ を使用しているコネクタに対応するバージョンに置き換えます。 最新の ​`pom.xml`​ ファイル情報を取得するには、​ [Anypoint Exchange](https://www.mulesoft.com/exchange/)​ でコネクタにアクセスし、​**\[Dependency Snippets (連動関係スニペット)\]**​ をクリックします。 ## [](#mule-プロジェクトで操作を設定する)Mule プロジェクトで操作を設定する Anypoint Studio で、次の手順を実行します。 1. コネクタがまだインストールされていない場合はインストールします。​[\[Install Anypoint MQ Connector\]](<#Install Anypoint MQ Connector>)​ を参照してください。 2. Mule プロジェクトを作成するか開きます。 3. **\[Mule Palette (Mule パレット)\]**​ で「mq」を検索します。 4. **\[Anypoint MQ\]**​ をクリックしてコネクタで使用可能な操作を表示します。 ![Anypoint MQ の \[Mule Palette (Mule パレット)\]](_images/amq-3x-connector-palette.png) 5. 操作を適宜設定します。
操作 説明

ACK

メッセージが適切にコンシュームされたことを示し、インフライト状況からメッセージを削除します。

Consume

キューからメッセージをコンシュームします。

NACK

メッセージの状況をインフライトからキュー待機中に変更し、再びサブスクライバーがコンシュームできるようにします。

Publish

宛先にメッセージをパブリッシュします。

Subscriber

キューのメッセージをリスンします。

### [](#デフォルトのサブスクライバーへの接続を設定する)デフォルトのサブスクライバーへの接続を設定する 各操作で、サブスクライバーへの接続を設定する必要があります。 Anypoint Studio で、次の手順を実行します。 1. **\[Mule Palette (Mule パレット)\]**​ からキャンバスに操作の名前をドラッグして選択します。 2. **\[Connector configuration (コネクタ設定)\]**​ の右にある緑のプラス記号をクリックします。 3. Anypoint Platform で、​**\[MQ\] > \[Destinations (宛先)\]**​ をクリックします。 4. 自分のリージョンの URL を Studio の ​**\[URL\]**​ 項目にコピーします。 5. Anypoint Platform で、​**\[MQ\] > \[Client Apps (クライアントアプリケーション)\]**​ をクリックします。 6. 必要に応じて、クライアントアプリケーションを作成します。 7. Anypoint Platform の ​**\[Client App ID (クライアントアプリケーション ID)\]**​ の値を Studio の ​**\[Client App ID (クライアントアプリケーション ID)\]**​ 項目にコピーします。 8. Anypoint Platform の ​**\[Client Secret (クライアントシークレット)\]**​ の値を Studio の ​**\[Client Secret (クライアントシークレット)\]**​ 項目にコピーします。 9. **\[Test Connection (接続をテスト)\]**​ をクリックして、Anypoint MQ のデフォルトのサブスクライバーを確認します。 10. **\[OK\]**​ をクリックします。 ## [](#関連情報)関連情報 - [チュートリアル: キューの作成とキューへのメッセージの送信](https://docs.mulesoft.com/jp/mq/mq-tutorial) - [Anypoint MQ Connector リファレンス](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-connector-reference) - [Anypoint MQ の概要](https://docs.mulesoft.com/jp/mq/) - [FAQ: Anypoint MQ](https://docs.mulesoft.com/jp/mq/mq-faq) - [Anypoint MQ の REST API](https://docs.mulesoft.com/jp/mq/mq-apis) - [Anypoint Exchange の Anypoint MQ Connector](https://anypoint.mulesoft.com/exchange/com.mulesoft.connectors/anypoint-mq-connector/) - [Anypoint MQ Connector Release Notes for Mule 4](https://docs.mulesoft.com/jp/release-notes/connector/anypoint-mq-connector-release-notes-mule-4) # Anypoint MQ Connector 3.x 移行ガイド Anypoint MQ Connector バージョン 3.0 には、簡素化された設定、標準化されたエンドポイント、非ブロック機能などの新機能があります。 ## [](#グローバル設定)グローバル設定 このコネクタのバージョン 2.x では、​`default-subscriber-config`​ に次のパラメーターが含まれています。 ```xml maxRedelivery="2" pollingTime="2000" fetchSize="5" fetchTimeout="5000" frequency="2000"> clientId="${client.id}" clientSecret="${client.secret}"/> circuitName="InvoiceSettings" onErrorTypes="JMS:TIMEOUT" errorsThreshold="10" tripTimeout="1" tripTimeoutUnit="MINUTES" /> ``` このコネクタのバージョン 3.x では、​`config`​ 要素に接続設定のみが含まれています。 ```xml ``` 次の表に、2.x の ​`config`​ 要素とそれに対応する 3.x の動作を示します。
要素 バージョン 2.x の動作 対応するバージョン 3.x の動作

(1)​ - 肯定応答

Anypoint MQ Subscriber ソースのグローバル肯定応答設定。

サブスクライバーインスタンスごとに設定します。

(2)​ - 最大再配信数

redeliveryCount​ ヘッダーにある Anypoint MQ メッセージの再配信の検索条件。

非推奨になり、スタンドアロン検索条件として使用できなくなりました。代わりに ​"DLQ + Idempotent Filter"​ パターンを使用します。

(3)​ - メッセージフェッチ

Anypoint MQ サービスからメッセージを取得する方法に関連するすべてのパラメーター (​prefetch​ 設定など)。

Subscriber Type​ パラメーターで各サブスクライバーのすべてのフェッチ戦略パラメーターを設定します。

(4)​ - 接続

接続パラメーター。

すべての接続設定で同じ設定です。

(5)​ - サーキットブレーカー

Anypoint MQ Subscriber ソースのサーキットブレーカー設定。

各サーキットブレーカーインスタンスを、各 Anypoint MQ サブスクライバーで参照されるスタンドアロン要素、またはこれを宣言する Subscriber ソースの非公開サーキットブレーカーとして処理します。

## [](#subscriber-サブスクライバー-リスナー)Subscriber (サブスクライバー) リスナー このコネクタのバージョン 3.x では、メッセージリスニングの ​`Subscriber`​ メッセージソース設定が変更されました。 新しいサブスクライバーについては、​[「Anypoint MQ Subscriber ソース」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-listener)​を参照してください。 ### [](#メッセージ肯定応答)メッセージ肯定応答 このコネクタのバージョン 2.x では、Subscriber ソースの肯定応答モードが設定レベルで定義されます。 ```xml ``` バージョン 3.x では、これらのパラメーターは Subscriber ソース自体に含まれ、その 1 つのサブスクライバーインスタンスにのみ影響します。肯定応答パラメーターのセマンティクスは、変更されていません。 Subscriber ソースの対応する設定を次に示します。 ```xml ``` ### [](#プリフェッチモード)プリフェッチモード このコネクタのバージョン 2.x では、​`prefetch`​ モードがデフォルトで有効になり、関連するすべてのパラメーターを設定レベルで指定していました。 ```xml ``` バージョン 3.x には、​`prefetch`​ 設定を制御する ​`Subscriber Type`​ という新しい概念があります。 ```xml ``` 次の表に、2.x の ​`prefetch`​ パラメーターとそれに対応する 3.x のパラメーターを示します。
バージョン 2.x のパラメーター 説明 対応するバージョン 3.x の動作 説明

fetchSize

config​ レベルで宣言されます。​fetchSize​ 値を 3 倍して計算されるローカルメッセージバッファのサイズを制御します。

maxLocalMessages

Subscriber​ レベルで宣言されます。ローカルメッセージバッファに含めることができる最大サイズを明示的に制御します。このパラメーターは、バッファをフルにするために必要な数のメッセージをフェッチするときの目標値として機能します。

Unresolved include directive in modules/ROOT/pages/anypoint-mq-new-3x-major.adoc - include::connectors::partial$mq-disclaimer.adoc[]

fetchTimeout

サービスからの応答を待機する時間を制御します。

非推奨

このパラメーターは使用できなくなりました。コネクタは、常にロングポーリングの最大値を使用します。

frequency

キューが空になった後のメッセージの再確認のタイミングを制御します。

非推奨

このパラメーターは使用できなくなりました。コネクタは、内部で頻度を処理します (1 秒で固定)。

pollingTime

fetchSize​ が ​0​ の場合にのみ使用されます。実質的には ​prefetch​ モードを無効にして ​polling​ モードを有効にします。

非推奨

このパラメーターは、より強力な ​polling​ subscriber type に置き換わっています。

### [](#ポーリングモード)ポーリングモード Anypoint MQ Connector のバージョン 2.x では、​`fetchSize`​ パラメーターを ​`0`​ に設定し、​`pollingTime`​ を固定頻度のポーリングスケジューラーとして使用することで ​`prefetch`​ モードを無効化できました。 このコネクタのバージョン 3.x では、Subscriber ソースのポーリングモードが簡素化および正規化されています。Mule Runtime Engine で標準で提供されるスケジューラーを ​`fixed-frequency`​ または ​`cron`​ として使用できます。 - `fixed-frequency` ```xml ``` - `cron` ```xml ``` 次の表に、2.x の ​`polling`​ パラメーターとそれに対応する 3.x のパラメーターを示します。
バージョン 2.x のパラメーター 説明 対応するバージョン 3.x の動作 説明

fetchSize

config​ レベルで宣言されます。​fetchSize​ 値を 3 倍して計算されるローカルメッセージバッファのサイズを制御します。

fetchSize

Subscriber​ レベルで宣言され、ポーリングの実行ごとにフェッチするメッセージの最大数 (1 ~ 10) を設定します。 デフォルトは 10 です。

fetchTimeout

サービスからの応答の待機時間を制限します。

非推奨

このパラメーターは使用できなくなりました。コネクタは、常にロングポーリングの最大値を使用します。

frequency

キューが空になった後のメッセージの再確認のタイミングを制御します。

非推奨

このパラメーターは使用できなくなりました。コネクタは、1 秒の一定間隔を使用します。

pollingTime

fetchSize​ が ​0​ の場合にのみ使用されます。実質的には ​prefetch​ モードを無効にして ​polling​ モードを有効にします。

非推奨

このパラメーターは、より強力な ​polling​ subscriber type に置き換わっています。

### [](#サーキットブレーカー)サーキットブレーカー サーキットブレーカーの宣言は ​`default-subscriber-config`​ で行われなくなり、グローバル要素宣言またはインライン宣言のいずれかになります。 このコネクタのバージョン 2.x では、サブスクライバー設定の一部としてサーキットブレーカーを宣言し、複数の Subscriber ソースで参照していました。 ```xml ``` このコネクタのバージョン 3.x では、1 つのスタンドアロングローバル要素を宣言し、各サブスクライバーからそのグローバル要素を参照します。接続をアプリケーションロジックサーキットにバインドすることはなくなりました。 ```xml ``` ## [](#publish-操作)Publish 操作 このコネクタのバージョン 2.x では、Publish 操作のプロパティは次のように宣言されます。 ```xml #[vars.messageBody] ``` バージョン 3.x では、同じプロパティを動的なマップとして宣言するため、変換されたり、メッセージごとにキーが固定されずにキー数が動的になったりする可能性があります。 ```xml #[vars.messageBody] ``` 次の表に、2.x の Publish 操作とそれに対応する 3.x の操作を示します。
変更 バージョン 2.x バージョン 3.x

実行種別

Blocking

Non-Blocking

プロパティパラメーター

固定されたキー - 値のマップ

動的な「コンテンツ」パラメーターのマップ

詳細は、​[「Anypoint MQ Publish 操作」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-publish)​を参照してください。 ## [](#consume-操作)Consume 操作 次の表に、2.x の Consume 操作とそれに対応する 3.x の操作を示します。
変更 バージョン 2.x バージョン 3.x

実行種別

Blocking

Non-Blocking

デフォルトの肯定応答モード

MANUAL

IMMEDIATE

詳細は、​[「Anypoint MQ Consume 操作」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-consume)​を参照してください。 ## [](#ack-および-nack-操作)ACK および NACK 操作 ACK および NACK 操作は、​`ackToken`​ 文字列を使用します。 `ackToken`​ は、​`attributes`​ 要素から取得できます。​`ackToken`​ 値は、​`MANUAL`​ 肯定応答モードのメッセージでのみ使用できます。 次の表に、2.x の ACK および NACK 操作とそれに対応する 3.x の操作を示します。
バージョン 2.x バージョン 3.x

messageContext​ パラメーターから受け取る操作の値は、特定のメッセージの ​attributes​ 要素である必要があります。

メッセージの ACK または NACK を実行するためのメッセージの ​ackToken​ 文字列。

詳細は、​[「Anypoint MQ ACK および NACK 操作」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack)​を参照してください。 ## [](#プラットフォームの互換性)プラットフォームの互換性
ソフトウェア バージョン

Mule Runtime Engine

4.1.1 以降

Anypoint Studio

v7 以降

# Anypoint MQ Connector のアップグレードおよび移行 最新の機能を使用するには、Anypoint MQ 用 Anypoint Connector (Anypoint MQ Connector) をバージョン 3.x からバージョン 4.x にアップグレードします。 ## [](#サポートされているアップグレードパス)サポートされているアップグレードパス
開始バージョン 終了バージョン

3.x

4.x

## [](#このリリースでの変更)このリリースでの変更 バージョン 4.x に含まれる一部の変更により、バージョン 3.x との後方互換性が失われます。 Anypoint MQ Connector 4.x には、次の機能のサポートが含まれています。 - クロスリージョンフェールオーバー Mule 4 用 Anypoint MQ Connector バージョン 4.0.7 以降では、標準キューのフェールオーバー機能が提供されます。 有効にすると、Mule アプリケーションがリージョン内の Anypoint MQ サーバーにアクセスできない場合、コネクタは代替リージョンの代替キューに切り替えて、メッセージをパブリッシュおよびコンシュームします。 詳細は、以下を参照してください。 - [mq::mq-failover.adoc](#mq::mq-failover.adoc) - [アプリケーションでのクロスリージョンフェールオーバーの設定](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-failover) - 接続アプリケーション Mule 4 用 Anypoint MQ Connector バージョン 4.0.1 以降には、​**\[Global Element Properties (グローバル要素のプロパティ)\]**​ ページの ​**\[URL\]**​ 項目の新しい形式が含まれています。 - Anypoint MQ Connector 3.x では、URL は次の形式でした。 `https://mq-us-east-1.anypoint.mulesoft.com/api/v1` - Anypoint MQ Connector 4.x では、URL 形式に ​`{orgID}`​ と ​`{envID}`​ が含まれます。次に例を示します。 `https://mq-us-east-1.anypoint.mulesoft.com/api/v1/organizations/{orgID}/environments/{envID}`

誤ったバージョンの URL を使用してコネクタを設定すると、 Anypoint MQ は次のエラーを返します。

Invalid api url provided​ (無効な API URL が指定されています)

フェールオーバー機能を使用するには、[アプリケーションでのクロスリージョンフェールオーバーの設定](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-failover)を参照してください。 ## [](#アップグレード前提条件)アップグレード前提条件 アップグレードを実行する前に、以前のバージョンに復元する必要がある場合に備えて、ファイル、データ、および設定のバックアップを作成します。 ## [](#アップグレード手順)アップグレード手順 Anypoint MQ Connector 4.x にアップグレードする手順は、次のとおりです。 1. Anypoint MQ の ​**\[Destinations (宛先)\]**​ ページで、​**\[Copy (コピー)\]**​ をクリックし、メニューから ​**\[Connector version 4.x and later (Connector バージョン 4.x 以降)\]**​ を選択します。 ![\[Copy (コピー)\] ボタンとコネクタバージョンオプションを含むメニュー](../../mq/_images/mq-tutorial-url.png) 2. Anypoint Studio で、Mule プロジェクトを開きます。 3. \[Mule Palette (Mule パレット)\] ビューで、​**\[Search in Exchange (Exchange 内を検索)\]**​ をクリックします。 4. **\[Add Dependencies to Project (連動関係をプロジェクトに追加)\]**​ で、検索項目に「​`Anypoint MQ`​」と入力します。 5. **\[Available modules (使用可能なモジュール)\]**​ で ​**\[Anypoint MQ Connector - Mule 4\]**​ を選択し、​**\[Add (追加)\]**​ をクリックします。 6. **\[Finish (完了)\]**​ をクリックします。 Anypoint Studio はコネクタを自動的にアップグレードします。 7. `pom.xml`​ ファイルで ​`mq-connector`​ 連動関係のバージョンが ​`4.0.5`​ 以降であることを確認します。 ```xml com.mulesoft.connectors anypoint-mq-connector 4.0.5 mule-plugin ``` 8. Anypoint Studio の ​**\[Anypoint MQ Config (Anypoint MQ 設定)\]**​ ウィンドウで、​**\[URL\]**​ 項目に URL を貼り付けます。 ![URL、クライアントアプリケーション ID、クライアントシークレットが表示されている \[Anypoint MQ Config (Anypoint MQ 設定)\] ページ](../../mq/_images/mq-tutorial-studio7-mq-config.png) 9. 認証方法に応じて、​**\[Client App ID (クライアントアプリケーション ID)\]**​ 項目と ​**\[Client Secret (クライアントシークレット)\]**​ 項目に入力します。 - **Connected App (接続アプリケーション)** **\[Anypoint Platform\]**​ > ​**\[Access Management (アクセス管理)\]**​ > ​**\[Connected Apps (接続アプリケーション)\]**​ から ​**\[ID\]**​ と ​**\[Secret (シークレット)\]**​ の値を貼り付けます。 Anypoint MQ での接続アプリケーションの設定についての詳細は、 [接続アプリケーションの設定](https://docs.mulesoft.com/jp/mq/mq-connected-apps) を参照してください。 - **Client App (クライアントアプリケーション)** **\[Anypoint Platform\]**​ > ​**\[MQ\]**​ > ​**\[Client Apps (クライアントアプリケーション)\]**​ から ​**\[Client App ID (クライアントアプリケーション ID)\]**​ と ​**\[Client Secret (クライアントシークレット)\]**​ の値を貼り付けます。 Anypoint MQ でのクライアントアプリケーションの設定についての詳細は、 [クライアントアプリケーションの設定](https://docs.mulesoft.com/jp/mq/mq-client-apps) を参照してください。 ## [](#アップグレード後の手順)アップグレード後の手順 コネクタの最新バージョンをインストールしたら、次の手順に従ってアップグレードを完了します。 1. Anypoint Studio で、​**\[Problems (問題)\]**​ ビューまたは ​**\[Console (コンソール)\]**​ ビューにエラーがないことを確認します。 2. プロジェクトの pom.xml を確認し、問題がないことを確認します。 3. 接続をテストして、操作が機能することを確認します。 ## [](#トラブルシューティング)トラブルシューティング パラメーターのキャッシュおよびメタデータのキャッシュで問題が発生した場合、Anypoint Studio を再起動しください。 ### [](#アップグレードの取り消し)アップグレードの取り消し 前のバージョンの Anypoint MQ Connector に戻すには、Anypoint Studio で、プロジェクトの ​`pom.xml`​ ファイル内の ​`mq-connector`​ 連動関係バージョンを前のバージョンに変更します。 ## [](#関連情報)関連情報 - [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors) - [MuleSoft ヘルプセンター](https://help.mulesoft.com) # Anypoint MQ Connector の概要 Anypoint MQ 用 Anypoint Connector (Anypoint MQ Connector) は、Anypoint MQ クラウドサービスを使用して Mule アプリケーションにパブリッシュ - サブスクライブメッセージングを提供します。 Anypoint MQ は、エンタープライズメッセージングパターン、マイクロサービス間の非同期通信、IoT などの広範なメッセージングのユースケースをサポートします。また、Anypoint MQ には、node.js、Java、Go、DevOps bash スクリプトなどのフレームワークを使用してメッセージのパブリッシュやコンシュームを行う REST API もあります。 ## [](#要件)要件 このコネクタを使用するには、Anypoint Studio、Anypoint Connector、アプリケーションフローに精通している必要があります。Anypoint MQ Connector は無料ですが、Anypoint Platform でサービスに接続するにはライセンスが必要です。 ### [](#互換性)互換性
ソフトウェア バージョン

Mule Runtime Engine

4.1.1 以降

Anypoint Studio

7 以降

このドキュメントは、Anypoint MQ Connector 4.x のみに適用されます。バージョン 3.x からバージョン 4.x に移行するには、 [Anypoint MQ Connector のアップグレードおよび移行](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-connector-upgrade-migrate) を参照してください。 ## [](#install-mq-connector)Anypoint MQ Connector をインストールする このコネクタは、Anypoint Studio で記述されたアプリケーション、スタンドアロンアプリケーション、または XML で記述されたアプリケーションで使用できます。 Anypoint Studio で、次の手順を実行します。 1. Mule プロジェクトを作成するか開きます。 2. Studio タスクバーの ​**Exchange**​ アイコンをクリックします。 3. Anypoint Platform のユーザー名とパスワードを入力し、​**\[Sign In (サインイン)\]**​ をクリックします。 4. **\[All assets (すべてのアセット)\]**​ を選択して「​`MQ`​」を検索します。 5. **\[Anypoint MQ Connector — Mule 4\]**​ を選択し、​**\[Add to project (プロジェクトに追加)\]**​ をクリックします。 6. 画面の指示に従ってコネクタをインストールします。 スタンドアロンアプリケーションまたは XML で記述されたアプリケーションの場合。 1. Anypoint Studio で Mule プロジェクトを開きます。 2. コネクタを連動関係として ​`pom.xml`​ ファイルに追加します。 ```xml com.mulesoft.connectors anypoint-mq-connector x.x.x mule-plugin ``` `x.x.x`​ を使用しているコネクタに対応するバージョンに置き換えます。 最新の ​`pom.xml`​ ファイル情報を取得するには、​ [Anypoint Exchange](https://www.mulesoft.com/exchange/)​ でコネクタにアクセスし、​**\[Dependency Snippets (連動関係スニペット)\]**​ をクリックします。 ## [](#mule-プロジェクトで操作を設定する)Mule プロジェクトで操作を設定する Anypoint Studio で、次の手順を実行します。 1. コネクタがまだインストールされていない場合はインストールします。 [Anypoint MQ Connector をインストールする](#install-mq-connector)​ を参照してください。 2. Mule プロジェクトを作成するか開きます。 3. **\[Mule Palette (Mule パレット)\]**​ で「​`mq`​」を検索します。 4. **\[Anypoint MQ\]**​ をクリックしてコネクタで使用可能な操作を表示します。 ![Anypoint MQ Connector が選択され、使用可能な操作が表示されている \[Mule Palette (Mule パレット)\]](_images/amq-4x-connector-palette.png) 5. 操作を適宜設定します。
操作 説明

ACK

メッセージが適切にコンシュームされたことを示し、​[In Flight (インフライト)]​ 状況からメッセージを削除します。

Consume

キューからメッセージをコンシュームします。

NACK

メッセージの状況を ​[In Flight (インフライト)]​ から ​[In Queue (キュー待機中)]​ に変更し、再びサブスクライバーがコンシュームできるようにします。

Publish

宛先にメッセージをパブリッシュします。

Subscriber

キューのメッセージをリスンします。

### [](#デフォルトのサブスクライバーへの接続を設定する)デフォルトのサブスクライバーへの接続を設定する 各操作で、サブスクライバーへの接続を設定する必要があります。 1. Anypoint Studio で、​**\[Mule Palette (Mule パレット)\]**​ からキャンバスに操作の名前をドラッグして選択します。 2. **\[Connector configuration (コネクタ設定)\]**​ の緑色のプラス記号をクリックします。 3. Anypoint Platform で、​**\[MQ\] > \[Destinations (宛先)\]**​ をクリックします。 4. **\[Copy (コピー)\]**​ をクリックし、メニューから ​**\[Connector version 4.x and later (Connector バージョン 4.x 以降)\]**​ を選択します。 ![\[Copy (コピー)\] ボタンとコネクタバージョンオプションを含むメニュー](../../mq/_images/mq-tutorial-url.png) 5. Anypoint Studio の ​**\[Anypoint MQ Config (Anypoint MQ 設定)\]**​ ウィンドウで、​**\[URL\]**​ 項目に URL を貼り付けます。 6. Anypoint MQ で、​**\[MQ\] > \[Client Apps (クライアントアプリケーション)\]**​ をクリックします。 必要に応じて、クライアントアプリケーションを作成します。 7. クライアントアプリケーション名をクリックして詳細ペインを表示します。 8. 詳細ペインのクライアントアプリケーション ID とクライアントシークレットの値を Studio の ​**\[Anypoint MQ Config (Anypoint MQ 設定)\]**​ ウィンドウにコピーします。 ![URL、クライアントアプリケーション ID、クライアントシークレットを含む \[Global Element Properties (グローバル要素のプロパティ)\]](../../mq/_images/mq-tutorial-studio7-mq-config.png) 9. **\[Test Connection (接続をテスト)\]**​ をクリックして、Anypoint MQ のデフォルトのサブスクライバーを確認します。 10. **\[OK\]**​ をクリックします。 詳細は、 ​[Anypoint MQ チュートリアル](https://docs.mulesoft.com/jp/mq/mq-tutorial)​の​[「Anypoint Studio 7 での Mule 4 アプリケーションの作成および Anypoint MQ への Mule 4 アプリケーションの接続」](https://docs.mulesoft.com/jp/mq/mq-tutorial#configure-mule-4-studio)​を参照してください。 ## [](#関連情報)関連情報 - [チュートリアル: キューの作成とキューへのメッセージの送信](https://docs.mulesoft.com/jp/mq/mq-tutorial) - [FAQ: Anypoint MQ](https://docs.mulesoft.com/jp/mq/mq-faq) - [Anypoint MQ の REST API](https://docs.mulesoft.com/jp/mq/mq-apis) - [Anypoint Exchange の Anypoint MQ Connector](https://anypoint.mulesoft.com/exchange/com.mulesoft.connectors/anypoint-mq-connector/) - [Anypoint MQ Connector Release Notes for Mule 4](https://docs.mulesoft.com/jp/release-notes/connector/anypoint-mq-connector-release-notes-mule-4) # Anypoint MQ Connector リファレンス サポートされるバージョン: Select Anypoint MQ Connector のバージョン 3.x Anypoint MQ は、顧客がアプリケーション間で高度な非同期メッセージングを実行できる、マルチテナントのクラウドメッセージングサービスです。 ## [](#設定)設定 * * * ### [](#config)設定 #### [](#パラメーター)パラメーター
名前 型 説明 デフォルト値 必須

Name (名前)

String (文字列)

この設定の名前を指定します。コネクタはこの名前を使用して設定を参照します。

x

Connection (接続)

この設定に指定する接続型を指定します。

x

Expiration Policy (有効期限ポリシー)

動的設定インスタンスがアイドル状態を続けられる最小時間を設定します。この時間が経過すると、Runtime でパージに相当するとみなされます。必要に応じて、期限切れのインスタンスがパージされてリソースが解放されます。

#### [](#接続種別)接続種別 ##### [](#config_connection)Connection (接続) ###### [](#パラメーター-2)パラメーター
名前 型 説明 デフォルト値 必須

URL

String (文字列)

[Anypoint Platform] > [MQ]​ コンソールからのキューのリージョン URL。

https://mq-us-east-1.anypoint.mulesoft.com/api/v1

Client App ID (クライアントアプリケーション ID)

String (文字列)

[Anypoint Platform] > [MQ] > [Client Apps (クライアントアプリケーション)]​ からの [Client App ID (クライアントアプリケーション ID)]。

x

Client Secret (クライアントシークレット)

String (文字列)

[Anypoint Platform] > [MQ] > [Client Apps (クライアントアプリケーション)]​ からの [Client Secret (クライアントシークレット)] 項目。

x

TLS Context

TLS

Host (ホスト)

String (文字列)

プロキシ要求の宛先ホスト。

Port (ポート)

Number (数値)

プロキシ要求の宛先ポート。

Username (ユーザー名)

String (文字列)

プロキシに対して認証するユーザー名。

Password (パスワード)

String (文字列)

プロキシに対して認証するパスワード。

Send Buffer Size (送信バッファサイズ)

Number (数値)

データ送信時に使用するバッファのサイズ (バイト) で、ソケット自体に設定されます。

Receive Buffer Size (受信バッファサイズ)

Number (数値)

データ受信時に使用するバッファのサイズ (バイト)。

Client Timeout (クライアントタイムアウト)

Number (数値)

ソケットの ​SO_TIMEOUT​ 値。ブロック操作で失敗するまでにソケットが待機する時間 (ミリ秒) を示します。 値が 0 (デフォルト) の場合、無限に待機します。

Send TCP With No Delay (遅延なく TCP を送信)

Boolean (ブール)

設定した場合、送信データはグループ化されず、即時に送信されます。 デフォルトは true です。

true

Linger (延長)

Number (数値)

SO_LINGER​ 値。これは、残りのデータが正しく送信されるように、ソケットが閉じるまでにかかる時間 (ミリ秒) を決定します。

Keep Alive (キープアライブ)

Boolean (ブール)

開いているソケットの ​SO_KEEPALIVE​ 動作。接続が使用できなくなった場合、開いていて長時間使用されていないソケット接続を自動的にチェックして閉じます。これはソケット自体のプロパティであり、サーバーへの接続がリサイクルされるまでキープアライブで維持されるかどうかを制御するためにサーバーソケットで使用します。

false

Connection Timeout (接続タイムアウト)

Number (数値)

タイムアウトで失敗する前に、リモートサーバーへのアウトバウンド接続が正常に作成されるまで待機するミリ秒数。デフォルトは 30 秒です。

30000

Reconnection (再接続)

アプリケーションがデプロイされると、すべてのコネクタで接続テストが実行されます。true に設定されている場合、各コネクタへのデプロイメント接続テストに失敗すると、デプロイメントが失敗します。

## [](#操作)操作 - [Ack](#ack) - [Consume](#consume) - [Nack](#nack) - [Publish](#publish) ### [](#関連付けられたソース)関連付けられたソース - [Subscriber](#subscriber) ### [](#ack)Ack `` メッセージが適切にコンシュームされたことを示す肯定応答を実行し、インフライト状況からメッセージを削除します。 #### [](#パラメーター-3)パラメーター
名前 型 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Ack Token (Ack トークン)

String (文字列)

MANUAL モードでのみ、手動 ACK および NACK 操作を実行するために使用される肯定応答トークン。

x

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]

#### [](#次の設定の場合)次の設定の場合 - [設定](#config) #### [](#スロー)スロー - ANYPOINT-MQ:ACKING - ANYPOINT-MQ:CONNECTIVITY - ANYPOINT-MQ:DESTINATION\_NOT\_FOUND - ANYPOINT-MQ:RETRY\_EXHAUSTED ### [](#consume)Consume `` #### [](#パラメーター-4)パラメーター
名前 型 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Queue (キュー)

String (文字列)

メッセージのフェッチ元であるキューまたはエクスチェンジの名前。

x

Acknowledgment Mode (肯定応答モード)

Enumeration (列挙)。次のいずれかになります。

  • IMMEDIATE

  • MANUAL

取得したメッセージに使用する肯定応答モード。

IMMEDIATE

肯定応答タイムアウト

Number (数値)

肯定応答または否定応答を待機しているコンシューマーによってメッセージが保持される期間。その期間が経過した後、任意のコンシューマーがメッセージを再び利用できるようになります。

0

Acknowledgment Timeout Unit (肯定応答タイムアウト単位)

Enumeration (列挙)。次のいずれかになります。

  • NANOSECONDS

  • MICROSECONDS

  • MILLISECONDS

  • SECONDS

  • MINUTES

  • HOURS

  • DAYS

acknowledgmentTimeout 設定に使用する時間単位。

MILLISECONDS

Polling Time (ポーリング時間)

Number (数値)

コンシュームの準備が完了する前にメッセージを待機する時間 (0-20000)。

  • この値が 0 より大きく、メッセージが到着しない場合、コネクタは ​ANYPOINT-MQ:TIMEOUT​ 例外をスローします。

  • この値が 0 でメッセージが到着しない場合、コネクタは ​NULL​ を返し、​ANYPOINT-MQ:TIMEOUT​ 例外をスローしません。

10000

Polling Time Unit (ポーリング時間単位)

Enumeration (列挙)。次のいずれかになります。

  • NANOSECONDS

  • MICROSECONDS

  • MILLISECONDS

  • SECONDS

  • MINUTES

  • HOURS

  • DAYS

pollingTime 設定に使用する時間単位。

MILLISECONDS

Output Mime Type (出力 MIME タイプ)

String (文字列)

操作の出力の MIME タイプ。

Output Encoding (出力エンコード)

String (文字列)

操作の出力のエンコード。

Streaming Strategy (ストリーミング戦略)

反復可能なストリームを使用します。

Target Variable (対象変数)

String (文字列)

操作の出力を保存する変数の名前。

Target Value (対象値)

String (文字列)

操作の出力に対して評価される式。この式の結果が対象変数に保存されます。

#[payload]

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]

#### [](#出力)出力

型

Binary (バイナリ)

Attributes Type (属性型)

#### [](#次の設定の場合-2)次の設定の場合 - [設定](#config) #### [](#スロー-2)スロー - ANYPOINT-MQ:ACKING - ANYPOINT-MQ:CONSUMING - ANYPOINT-MQ:CONNECTIVITY - ANYPOINT-MQ:TIMEOUT - ANYPOINT-MQ:DESTINATION\_NOT\_FOUND - ANYPOINT-MQ:RETRY\_EXHAUSTED ### [](#nack)Nack `` 否定応答を実行し、メッセージの状況を「In Flight (インフライト)」から「In Queue (キュー待機中)」に変更し、サブスクライバーがコンシュームできるようにします。 #### [](#パラメーター-5)パラメーター
名前 型 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Ack Token (Ack トークン)

String (文字列)

MANUAL モードでのみ、手動 ACK および NACK 操作を実行するために使用される肯定応答トークン。

x

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]

#### [](#次の設定の場合-3)次の設定の場合 - [設定](#config) #### [](#スロー-3)スロー - ANYPOINT-MQ:NACKING - ANYPOINT-MQ:CONNECTIVITY - ANYPOINT-MQ:DESTINATION\_NOT\_FOUND - ANYPOINT-MQ:RETRY\_EXHAUSTED ### [](#publish)Publish `` #### [](#パラメーター-6)パラメーター
名前 型 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Destination (宛先)

String (文字列)

メッセージのフェッチ元であるキューまたはエクスチェンジの名前。

x

Output Mime Type (出力 MIME タイプ)

String (文字列)

操作の出力の MIME タイプ。

Output Encoding (出力エンコード)

String (文字列)

操作の出力のエンコード。

Streaming Strategy (ストリーミング戦略)

反復可能なストリームを使用します。

Body (本文)

Binary (バイナリ)

送信メッセージの本文。JSON メッセージの一部の文字列として送信されます。パブリッシュされるコンテンツは、文字列形式の逐次化と並列化を実行できる必要があります。文字列が適切に逐次化されていないオブジェクトを送信すると、受信側でメッセージが読めなくなります。

#[payload]

User Properties (ユーザープロパティ)

Object (オブジェクト)

メッセージに設定されたカスタムユーザープロパティ。

Message Id (メッセージ ID)

String (文字列)

送信メッセージの有効なカスタムユーザー ID。指定されていない場合、ブローカーによって作成されます。

Message Group Id (メッセージグループ ID)

String (文字列)

パブリッシュされたメッセージが属するメッセージグループの ID。FIFO キューでのみ指定します。

メッセージグループ ID には最大 128 文字の英数字 (a-z, A-Z、0-9) および句読点 (! " # $ % & ' ( ) * + , - . / : ; < = > ? @ [ ] \ ^ _ ` { } | ~) を含めることができます。

Delivery Delay (配信遅延)

Number (数値)

コンシュームする前にメッセージの配信を延期する時間 (0 ~ 15 分)。

MILLISECONDS

Delivery Delay Unit (配信遅延単位)

Enumeration (列挙)

Delivery Delay​ パラメーターで使用する時間単位。

  • MILLISECONDS

  • SECONDS

  • MINUTES

MILLISECONDS

Send Content-Type (コンテンツタイプの送信)

Boolean (ブール)

本文種別が Anypoint MQ メッセージプロパティとして送信される場合は true に設定します。

true

Target Variable (対象変数)

String (文字列)

操作の出力を保存する変数の名前。

Target Value (対象値)

String (文字列)

操作の出力に対して評価する式。式の結果は対象変数に保存されます。

#[payload]

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]

#### [](#出力-2)出力

型

Binary (バイナリ)

Attributes Type (属性型)

#### [](#次の設定の場合-4)次の設定の場合 - [設定](#config) #### [](#スロー-4)スロー - ANYPOINT-MQ:PUBLISHING - ANYPOINT-MQ:CONNECTIVITY - ANYPOINT-MQ:ILLEGAL\_BODY - ANYPOINT-MQ:DESTINATION\_NOT\_FOUND - ANYPOINT-MQ:RETRY\_EXHAUSTED ## [](#ソース)ソース ### [](#subscriber)Subscriber `` Anypoint MQ Subscriber メッセージソースは、指定された宛先名からメッセージを取得します。 #### [](#パラメーター-7)パラメーター
名前 型 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Queue (キュー)

String (文字列)

メッセージの取得元のキューの名前。

x

Subscriber Type (サブスクライバー種別)

次のいずれかになります。

サービスからメッセージにサブスクライブするときに使用する戦略。

Acknowledgment Mode (肯定応答モード)

Enumeration (列挙)。次のいずれかになります。

  • IMMEDIATE

  • AUTO

  • MANUAL

取得したメッセージに使用する肯定応答モード。

AUTO

肯定応答タイムアウト

Number (数値)

サブスクライバーが使用可能になる前に、肯定応答または否定応答を待機しているブローカーによってメッセージが保持される期間。上書きされていない場合のすべての Subscriber ソースのデフォルトの ​acknowledgmentTimeout​ 値を指定します。 このパラメーターの最大値は 12 時間です。

acknowledgementTimeout​ が設定されていない場合 (デフォルト)、コネクタでは少なくとも 2 分の TTL 値が使用されます。

Acknowledgment Timeout Unit (肯定応答タイムアウト単位)

Enumeration (列挙)。次のいずれかになります。

  • NANOSECONDS

  • MICROSECONDS

  • MILLISECONDS

  • SECONDS

  • MINUTES

  • HOURS

  • DAYS

acknowledgmentTimeout 設定に使用する時間単位。

MILLISECONDS

サーキットブレーカー

Output Mime Type (出力 MIME タイプ)

String (文字列)

操作の出力の MIME タイプ。

Output Encoding (出力エンコード)

String (文字列)

操作の出力のエンコード。

Primary Node Only (プライマリノードのみ)

Boolean (ブール)

クラスターでの実行時、このソースをプライマリノード上でのみ実行する必要があるかどうか。

Streaming Strategy (ストリーミング戦略)

反復可能なストリームを使用します。

Redelivery Policy (再配信ポリシー)

メッセージ再配信ポリシー。

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]

#### [](#出力-3)出力

型

Binary (バイナリ)

Attributes Type (属性型)

#### [](#次の設定の場合-5)次の設定の場合 - [設定](#config) ## [](#型)型 ### [](#Tls)TLS
項目 型 説明 デフォルト値 必須

Enabled Protocols (有効化されたプロトコル)

String (文字列)

このコンテキストで有効化されたプロトコルのカンマ区切りリスト。

Enabled Cipher Suites (有効化された暗号化スイート)

String (文字列)

このコンテキストで有効化された暗号化スイートのカンマ区切りリスト。

トラストストア

Key Store (キーストア)

Revocation Check (取り消し確認)

### [](#TrustStore)トラストストア
項目 型 説明 デフォルト値 必須

Path (パス)

String (文字列)

トラストストアの場所。可能であれば、現在のクラスパスとファイルシステムに対して解決されます。

Password (パスワード)

String (文字列)

トラストストアを保護するために使用するパスワード。

Type (型)

String (文字列)

使用するストアの型。

Algorithm (アルゴリズム)

String (文字列)

トラストストアで使用するアルゴリズム。

Insecure (安全ではない)

Boolean (ブール)

true の場合、証明書の検証が行われず、攻撃に対して脆弱な接続がレンダリングされます。各自の責任で使用してください。

### [](#KeyStore)Key Store (キーストア)
項目 型 説明 デフォルト値 必須

Path (パス)

String (文字列)

キーストアの場所。可能であれば、現在のクラスパスとファイルシステムに対して解決されます。

Type (型)

String (文字列)

使用するストアの型。

Alias (別名)

String (文字列)

キーストアに多数の非公開キーがある場合、使用する鍵の別名を示します。定義しない場合、ファイルにある最初のキーがデフォルトで使用されます。

Key Password (キーパスワード)

String (文字列)

非公開キーを保護するために使用するパスワード。

Password (パスワード)

String (文字列)

キーストアを保護するために使用するパスワード。

Algorithm (アルゴリズム)

String (文字列)

キーストアで使用するアルゴリズム。

### [](#standard-revocation-check)Standard Revocation Check (標準の取り消し確認)
項目 型 説明 デフォルト値 必須

Only End Entities (最後のエンティティのみ)

Boolean (ブール)

証明書チェーンの最後の要素のみを確認します。

Prefer Crls (CRL を優先)

Boolean (ブール)

OCSP 証明証明書を使用する前に、CRL 署名証明書を使用します。

No Fallback (代替なし)

Boolean (ブール)

第 2 の確認方法 ([Prefer Crls (CRL を優先)] で選択されていない方法) を使用しません。

Soft Fail (ソフトフェール)

Boolean (ブール)

取り消しサーバーに到達できない場合や取り消しサーバーが混雑している場合に検証エラーを回避します。

### [](#custom-ocsp-responder)Custom OCSP Responder (カスタム OCSP レスポンダー)
項目 型 説明 デフォルト値 必須

Url

String (文字列)

OCSP レスポンダーの URL。

Cert Alias (証明書の別名)

String (文字列)

OCSP 応答の署名証明書の別名 (存在する場合)。署名証明書がトラストストア内にある必要があります。

### [](#crl-file)CRL File (CRL ファイル)
項目 型 説明 デフォルト値 必須

Path (パス)

String (文字列)

CRL ファイルへのパス。

### [](#Reconnection)Reconnection (再接続)
項目 型 説明 デフォルト値 必須

Fails Deployment (デプロイメントに失敗)

Boolean (ブール)

アプリケーションがデプロイされると、すべてのコネクタで接続テストが実行されます。true に設定されている場合、各コネクタへのデプロイメント接続テストに失敗すると、デプロイメントが失敗します。

Reconnection Strategy (再接続戦略)

使用する再接続戦略。

Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]

### [](#reconnect)Reconnect (再接続)
Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]
項目 型 説明 デフォルト値 必須

Frequency (頻度)

Number (数値)

再接続する頻度 (ミリ秒)。

Count (数)

Number (数値)

再接続を試みる回数

blocking (ブロック)

Boolean (ブール)

false の場合、再接続戦略が個別の非ブロックスレッドで実行されます。

true

### [](#reconnect-forever)Reconnect Forever (繰り返し再接続)
Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]
項目 型 説明 デフォルト値 必須

Frequency (頻度)

Number (数値)

再接続する頻度 (ミリ秒)。

blocking (ブロック)

Boolean (ブール)

false の場合、再接続戦略が個別の非ブロックスレッドで実行されます。

true

### [](#ExpirationPolicy)Expiration Policy (有効期限ポリシー)
項目 型 説明 デフォルト値 必須

Max Idle Time (最大アイドル時間)

Number (数値)

有効期限の対象とみなされるまで、動的設定インスタンスがアイドル状態を維持できる最大時間のスカラー時間値

Time Unit (時間単位)

Enumeration (列挙)。次のいずれかになります。

  • NANOSECONDS

  • MICROSECONDS

  • MILLISECONDS

  • SECONDS

  • MINUTES

  • HOURS

  • DAYS

maxIdleTime 属性の時間単位

### [](#AnypointMQMessageAttributes)Anypoint MQ メッセージの属性
項目 型 説明 デフォルト値 必須

Destination (宛先)

String (文字列)

メッセージの宛先。

x

Headers (ヘッダー)

Object (オブジェクト)

メッセージヘッダー。

x

Properties (プロパティ)

Object (オブジェクト)

メッセージプロパティ。

x

Message Id (メッセージ ID)

String (文字列)

メッセージの ID。

x

Delivery Count (配信数)

Number (数値)

メッセージが配信された回数。

x

Content Type (コンテンツタイプ)

String (文字列)

メッセージ本文のコンテンツタイプ。

Ack Token (Ack トークン)

String (文字列)

MANUAL モードでのみ、手動 ACK および NACK 操作を実行するために使用される肯定応答トークン。

### [](#circuitBreaker)サーキットブレーカー
項目 型 説明 デフォルト値 必須

On Error Types (エラー時種別)

String (文字列)

フローの実行中に発生したサーキットの障害と同等のエラー種別。デフォルトでは、すべてのエラーが障害とみなされます。

Errors Threshold (エラーしきい値)

Number (数値)

サーキットで考慮される、サーキットがオープンになるのに発生する必要があるエラーの回数。

x

Trip Timeout (トリップタイムアウト)

Number (数値)

errorsThreshold​ に達してからサーキットがオープンのままである期間。

x

Trip Timeout Unit (トリップタイムアウト単位)

Enumeration (列挙)。次のいずれかになります。

  • NANOSECONDS

  • MICROSECONDS

  • MILLISECONDS

  • SECONDS

  • MINUTES

  • HOURS

  • DAYS

tripTimeout 値にバインドされる時間単位。

MILLISECONDS

### [](#repeatable-in-memory-stream)Repeatable In Memory Stream (反復可能なメモリ内ストリーム)
項目 型 説明 デフォルト値 必須

Initial Buffer Size (初期バッファサイズ)

Number (数値)

ストリームをコンシュームしてランダムにアクセスするために割り当てられるメモリ量。ストリームのデータ量がこのバッファサイズを超える場合は、bufferSizeIncrement 属性に従って、maxInMemorySize を上限としてバッファが拡張されます。

Buffer Size Increment (バッファサイズ増分)

Number (数値)

バッファサイズをどれだけ拡張できるか。0 以下に設定した場合、バッファがフルになると ​STREAM_MAXIMUM_SIZE_EXCEEDED​ が発生します。

Max Buffer Size (最大バッファサイズ)

Number (数値)

STREAM_MAXIMUM_SIZE_EXCEEDED​ が発生する前に使用できるメモリの最大量。0 以下に設定した場合、メモリは無制限になります。

Buffer Unit (バッファ単位)

Enumeration (列挙)。次のいずれかになります。

  • BYTE (バイト)

  • KB

  • MB

  • GB

これらのすべての属性の単位

### [](#repeatable-file-store-stream)Repeatable File Store Stream (反復可能なファイルストアストリーム)
項目 型 説明 デフォルト値 必須

Max In Memory Size (最大メモリ内サイズ)

Number (数値)

コンテンツがディスクにバッファされる前にストリームがローカルに保持できるメモリの最大量。

Buffer Unit (バッファ単位)

Enumeration (列挙)。次のいずれかになります。

  • BYTE (バイト)

  • KB

  • MB

  • GB

maxInMemorySize で表される単位。

### [](#RedeliveryPolicy)Redelivery Policy (再配信ポリシー) 再配信ポリシーについての詳細は、​[「再配信ポリシーの設定」](https://docs.mulesoft.com/mule-runtime/latest/redelivery-policy)​を参照してください。 ### [](#AnypointMQMessagePublishAttributes)Anypoint MQ Message Publish Attributes (Anypoint MQ メッセージのパブリッシュ属性)
項目 型 説明 デフォルト値 必須

Message Id (メッセージ ID)

String (文字列)

パブリッシュされたメッセージの ID。

x

### [](#prefetch)プリフェッチ
項目 型 説明 デフォルト値 必須

Max Local Messages (最大ローカルメッセージ数)

Number (数値)

サービスからプリフェッチされ、フローにディスパッチされる前にローカルに保持されるメッセージの最大数。

Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]

Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]

30

### [](#polling)ポーリング
項目 型 説明 デフォルト値 必須

Fetch size (フェッチサイズ)

Number (数値)

ポーリングの実行ごとにフェッチするメッセージの最大数 (1 ~ 10)。

10 よりも大きい値を設定すると、エラーが発生します。

10

Scheduling Strategy (スケジュール戦略)

scheduling-strategy

サービスからのメッセージのフェッチをトリガーするスケジュール戦略。

x

### [](#fixed-frequency)一定間隔
項目 型 説明 デフォルト値 必須

Frequency (頻度)

Number (数値)

Polling frequency (ポーリング頻度)

1000ms

(1 秒)

Start delay (開始遅延)

Number (数値)

開始前にスケジューラーが待機する時間。

10

Time unit (時間単位)

Enumeration (列挙)。次のいずれかになります。

  • DAYS

  • HOURS

  • MINUTES

  • SECONDS

  • MILLISECONDS

startDelay​ 設定で使用する時間単位。

MILLISECONDS

### [](#cron)Cron
項目 型 説明 デフォルト値 必須

Expression (式)

Cron 式

x

Time Zone (タイムゾーン)

String (文字列)

式が基づくタイムゾーンの ID。

## [](#関連情報)関連情報 - [一定間隔](https://docs.mulesoft.com/mule-runtime/latest/scheduler-concept#fixed-frequency) - [Cron 式](https://docs.mulesoft.com/mule-runtime/latest/scheduler-concept#cron-expressions) # Anypoint MQ Connector リファレンス サポートされるバージョン: Select Anypoint MQ Connector のバージョン 4.x Anypoint MQ は、顧客がアプリケーション間で高度な非同期メッセージングを実行できる、マルチテナントのクラウドメッセージングサービスです。 ## [](#設定)設定 * * * ### [](#config)設定 #### [](#パラメーター)パラメーター
名前 型 説明 デフォルト値 必須

Name (名前)

String (文字列)

この設定の名前を指定します。コネクタはこの名前を使用して設定を参照します。

x

Connection (接続)

この設定に指定する接続型を指定します。

x

Expiration Policy (有効期限ポリシー)

動的設定インスタンスがアイドル状態を続けられる最小時間を設定します。この時間が経過すると、Runtime でパージに相当するとみなされます。必要に応じて、期限切れのインスタンスがパージされてリソースが解放されます。

#### [](#接続種別)接続種別 ##### [](#config_connection)Connection (接続) ###### [](#パラメーター-2)パラメーター
名前 型 説明 デフォルト値 必須

URL

String (文字列)

[Anypoint Platform] > [MQ]​ コンソールからのキューのリージョン URL。

https://mq-us-east-1.anypoint.mulesoft.com/api/v1

Client App ID (クライアントアプリケーション ID)

String (文字列)

次のいずれかの値。

  • [Anypoint Platform]​ > ​[Access Management (アクセス管理)]​ > ​[Connected Apps (接続アプリケーション)]​ からの ​[ID]​

  • [Anypoint Platform] > [MQ] > [Client Apps (クライアントアプリケーション)]​ からの ​[Client App ID (クライアントアプリケーション ID)]

x

Client Secret (クライアントシークレット)

String (文字列)

次のいずれかの値。

  • [Anypoint Platform]​ > ​[Access Management (アクセス管理)]​ > ​[Connected Apps (接続アプリケーション)]​ からの ​[Secret (シークレット)]​

  • [Anypoint Platform] > [MQ] > [Client Apps (クライアントアプリケーション)]​ からの ​[Client Secret (クライアントシークレット)]

x

TLS Context

TLS

Host (ホスト)

String (文字列)

プロキシ要求の宛先ホスト。

Port (ポート)

Number (数値)

プロキシ要求の宛先ポート。

Username (ユーザー名)

String (文字列)

プロキシに対して認証するユーザー名。

Password (パスワード)

String (文字列)

プロキシに対して認証するパスワード。

Send Buffer Size (送信バッファサイズ)

Number (数値)

データ送信時に使用するバッファのサイズ (バイト) で、ソケット自体に設定されます。

Receive Buffer Size (受信バッファサイズ)

Number (数値)

データ受信時に使用するバッファのサイズ (バイト)。

Client Timeout (クライアントタイムアウト)

Number (数値)

ソケットの ​SO_TIMEOUT​ 値。ブロック操作で失敗するまでにソケットが待機する時間 (ミリ秒) を示します。 値が 0 (デフォルト) の場合、無限に待機します。

Send TCP With No Delay (遅延なく TCP を送信)

Boolean (ブール)

設定した場合、送信データはグループ化されず、即時に送信されます。 デフォルトは true です。

true

Linger (延長)

Number (数値)

SO_LINGER​ 値。これは、残りのデータが正しく送信されるように、ソケットが閉じるまでにかかる時間 (ミリ秒) を決定します。

Keep Alive (キープアライブ)

Boolean (ブール)

開いているソケットの ​SO_KEEPALIVE​ 動作。接続が使用できなくなった場合、開いていて長時間使用されていないソケット接続を自動的にチェックして閉じます。これはソケット自体のプロパティであり、サーバーへの接続がリサイクルされるまでキープアライブで維持されるかどうかを制御するためにサーバーソケットで使用します。

false

Connection Timeout (接続タイムアウト)

Number (数値)

タイムアウトで失敗する前に、リモートサーバーへのアウトバウンド接続が正常に作成されるまで待機するミリ秒数。デフォルトは 30 秒です。

30000

Reconnection (再接続)

アプリケーションがデプロイされると、すべてのコネクタで接続テストが実行されます。true に設定されている場合、各コネクタへのデプロイメント接続テストに失敗すると、デプロイメントが失敗します。

## [](#操作)操作 - [Ack](#ack) - [Consume](#consume) - [Nack](#nack) - [Publish](#publish) ### [](#関連付けられたソース)関連付けられたソース - [Subscriber](#subscriber) ### [](#ack)Ack `` メッセージが適切にコンシュームされたことを示す肯定応答を実行し、インフライト状況からメッセージを削除します。 #### [](#パラメーター-3)パラメーター
名前 型 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Ack Token (Ack トークン)

String (文字列)

MANUAL モードでのみ、手動 ACK および NACK 操作を実行するために使用される肯定応答トークン。

x

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

#### [](#次の設定の場合)次の設定の場合 - [設定](#config) #### [](#スロー)スロー - ANYPOINT-MQ:ACKING - ANYPOINT-MQ:CONNECTIVITY - ANYPOINT-MQ:DESTINATION\_NOT\_FOUND - ANYPOINT-MQ:RETRY\_EXHAUSTED ### [](#consume)Consume `` #### [](#パラメーター-4)パラメーター
名前 型 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Queue (キュー)

String (文字列)

メッセージのフェッチ元であるキューまたはエクスチェンジの名前。

x

Acknowledgment Mode (肯定応答モード)

Enumeration (列挙)。次のいずれかになります。

  • IMMEDIATE

  • MANUAL

取得したメッセージに使用する肯定応答モード。

IMMEDIATE

肯定応答タイムアウト

Number (数値)

肯定応答または否定応答を待機しているコンシューマーによってメッセージが保持される期間。その期間が経過した後、任意のコンシューマーがメッセージを再び利用できるようになります。

0

Acknowledgment Timeout Unit (肯定応答タイムアウト単位)

Enumeration (列挙)。次のいずれかになります。

  • NANOSECONDS (ナノ秒)

  • MICROSECONDS (マイクロ秒)

  • MILLISECONDS (ミリ秒)

  • SECONDS (秒)

  • MINUTES (分)

  • HOURS (時間)

  • DAYS (日)

acknowledgmentTimeout 設定に使用する時間単位。

MILLISECONDS

Polling Time (ポーリング時間)

Number (数値)

コンシュームの準備が完了する前にメッセージを待機する時間 (0-20000)。

  • この値が 0 より大きく、メッセージが到着しない場合、コネクタは ​ANYPOINT-MQ:TIMEOUT​ 例外をスローします。

  • この値が 0 でメッセージが到着しない場合、コネクタは ​NULL​ を返し、​ANYPOINT-MQ:TIMEOUT​ 例外をスローしません。

10000

Polling Time Unit (ポーリング時間単位)

Enumeration (列挙)。次のいずれかになります。

  • NANOSECONDS (ナノ秒)

  • MICROSECONDS (マイクロ秒)

  • MILLISECONDS (ミリ秒)

  • SECONDS (秒)

  • MINUTES (分)

  • HOURS (時間)

  • DAYS (日)

pollingTime 設定に使用する時間単位。

MILLISECONDS

Output Mime Type (出力 MIME タイプ)

String (文字列)

操作の出力の MIME タイプ。

Output Encoding (出力エンコード)

String (文字列)

操作の出力のエンコード。

Streaming Strategy (ストリーミング戦略)

反復可能なストリームを使用します。

Target Variable (対象変数)

String (文字列)

操作の出力を保存する変数の名前。

Target Value (対象値)

String (文字列)

操作の出力に対して評価される式。この式の結果が対象変数に保存されます。

#[payload]

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

#### [](#出力)出力

型

Binary (バイナリ)

Attributes Type (属性型)

#### [](#次の設定の場合-2)次の設定の場合 - [設定](#config) #### [](#スロー-2)スロー - ANYPOINT-MQ:ACKING - ANYPOINT-MQ:CONSUMING - ANYPOINT-MQ:CONNECTIVITY - ANYPOINT-MQ:TIMEOUT - ANYPOINT-MQ:DESTINATION\_NOT\_FOUND - ANYPOINT-MQ:RETRY\_EXHAUSTED ### [](#nack)Nack `` 否定応答を実行し、メッセージの状況を「In Flight (インフライト)」から「In Queue (キュー待機中)」に変更し、サブスクライバーがコンシュームできるようにします。 #### [](#パラメーター-5)パラメーター
名前 型 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Ack Token (Ack トークン)

String (文字列)

MANUAL モードでのみ、手動 ACK および NACK 操作を実行するために使用される肯定応答トークン。

x

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

#### [](#次の設定の場合-3)次の設定の場合 - [設定](#config) #### [](#スロー-3)スロー - ANYPOINT-MQ:NACKING - ANYPOINT-MQ:CONNECTIVITY - ANYPOINT-MQ:DESTINATION\_NOT\_FOUND - ANYPOINT-MQ:RETRY\_EXHAUSTED ### [](#publish)Publish `` #### [](#パラメーター-6)パラメーター
名前 型 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Destination (宛先)

String (文字列)

メッセージのフェッチ元であるキューまたはエクスチェンジの名前。

x

Output Mime Type (出力 MIME タイプ)

String (文字列)

操作の出力の MIME タイプ。

Output Encoding (出力エンコード)

String (文字列)

操作の出力のエンコード。

Streaming Strategy (ストリーミング戦略)

反復可能なストリームを使用します。

Body (本文)

Binary (バイナリ)

送信メッセージの本文。JSON メッセージの一部の文字列として送信されます。パブリッシュされるコンテンツは、文字列形式の逐次化と並列化を実行できる必要があります。文字列が適切に逐次化されていないオブジェクトを送信すると、受信側でメッセージが読めなくなります。

#[payload]

User Properties (ユーザープロパティ)

Object (オブジェクト)

メッセージに設定されたカスタムユーザープロパティ。

Message Id (メッセージ ID)

String (文字列)

送信メッセージの有効なカスタムユーザー ID。指定されていない場合、ブローカーによって作成されます。

Message Group Id (メッセージグループ ID)

String (文字列)

パブリッシュされたメッセージが属するメッセージグループの ID。FIFO キューでのみ指定します。

メッセージグループ ID には最大 128 文字の英数字 (a-z, A-Z、0-9) および句読点 (! " # $ % & ' ( ) * + , - . / : ; < = > ? @ [ ] \ ^ _ ` { } | ~) を含めることができます。

Delivery Delay (配信遅延)

Number (数値)

コンシュームする前にメッセージの配信を延期する時間 (0 ~ 15 分)。

MILLISECONDS

Delivery Delay Unit (配信遅延単位)

Enumeration (列挙)

Delivery Delay​ パラメーターで使用する時間単位。

  • MILLISECONDS

  • SECONDS

  • MINUTES

MILLISECONDS

Send Content-Type (コンテンツタイプの送信)

Boolean (ブール)

本文種別が Anypoint MQ メッセージプロパティとして送信される場合は true に設定します。

true

Target Variable (対象変数)

String (文字列)

操作の出力を保存する変数の名前。

Target Value (対象値)

String (文字列)

操作の出力に対して評価する式。式の結果は対象変数に保存されます。

#[payload]

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

#### [](#出力-2)出力

型

Binary (バイナリ)

Attributes Type (属性型)

#### [](#次の設定の場合-4)次の設定の場合 - [設定](#config) #### [](#スロー-4)スロー - ANYPOINT-MQ:PUBLISHING - ANYPOINT-MQ:CONNECTIVITY - ANYPOINT-MQ:ILLEGAL\_BODY - ANYPOINT-MQ:DESTINATION\_NOT\_FOUND - ANYPOINT-MQ:RETRY\_EXHAUSTED ## [](#ソース)ソース ### [](#subscriber)Subscriber `` Anypoint MQ Subscriber メッセージソースは、指定された宛先名からメッセージを取得します。 #### [](#パラメーター-7)パラメーター
名前 型 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Queue (キュー)

String (文字列)

メッセージの取得元のキューの名前。

x

Subscriber Type (サブスクライバー種別)

次のいずれかになります。

サービスからメッセージにサブスクライブするときに使用する戦略。

Acknowledgment Mode (肯定応答モード)

Enumeration (列挙)。次のいずれかになります。

  • IMMEDIATE

  • AUTO

  • MANUAL

取得したメッセージに使用する肯定応答モード。

AUTO

肯定応答タイムアウト

Number (数値)

サブスクライバーが使用可能になる前に、肯定応答または否定応答を待機しているブローカーによってメッセージが保持される期間。上書きされていない場合のすべての Subscriber ソースのデフォルトの ​acknowledgmentTimeout​ 値を指定します。このパラメーターの最大値は 12 時間です。

acknowledgementTimeout​ が設定されていない場合 (デフォルト)、コネクタでは少なくとも 2 分の TTL 値が使用されます。

Acknowledgment Timeout Unit (肯定応答タイムアウト単位)

Enumeration (列挙)。次のいずれかになります。

  • NANOSECONDS (ナノ秒)

  • MICROSECONDS (マイクロ秒)

  • MILLISECONDS (ミリ秒)

  • SECONDS (秒)

  • MINUTES (分)

  • HOURS (時間)

  • DAYS (日)

acknowledgmentTimeout 設定に使用する時間単位。

MILLISECONDS

サーキットブレーカー

Output Mime Type (出力 MIME タイプ)

String (文字列)

操作の出力の MIME タイプ。

Output Encoding (出力エンコード)

String (文字列)

操作の出力のエンコード。

Primary Node Only (プライマリノードのみ)

Boolean (ブール)

クラスターでの実行時、このソースをプライマリノード上でのみ実行する必要があるかどうか。

Streaming Strategy (ストリーミング戦略)

反復可能なストリームを使用します。

Redelivery Policy (再配信ポリシー)

メッセージ再配信ポリシー。

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

#### [](#出力-3)出力

型

Binary (バイナリ)

Attributes Type (属性型)

#### [](#次の設定の場合-5)次の設定の場合 - [設定](#config) ## [](#型)型 ### [](#Tls)TLS
項目 型 説明 デフォルト値 必須

Enabled Protocols (有効化されたプロトコル)

String (文字列)

このコンテキストで有効化されたプロトコルのカンマ区切りリスト。

Enabled Cipher Suites (有効化された暗号化スイート)

String (文字列)

このコンテキストで有効化された暗号化スイートのカンマ区切りリスト。

トラストストア

Key Store (キーストア)

Revocation Check (取り消し確認)

### [](#TrustStore)トラストストア
項目 型 説明 デフォルト値 必須

Path (パス)

String (文字列)

トラストストアの場所。可能であれば、現在のクラスパスとファイルシステムに対して解決されます。

Password (パスワード)

String (文字列)

トラストストアを保護するために使用するパスワード。

Type (型)

String (文字列)

使用するストアの型。

Algorithm (アルゴリズム)

String (文字列)

トラストストアで使用するアルゴリズム。

Insecure (安全ではない)

Boolean (ブール)

true の場合、証明書の検証が行われず、攻撃に対して脆弱な接続がレンダリングされます。各自の責任で使用してください。

### [](#KeyStore)Key Store (キーストア)
項目 型 説明 デフォルト値 必須

Path (パス)

String (文字列)

キーストアの場所。可能であれば、現在のクラスパスとファイルシステムに対して解決されます。

Type (型)

String (文字列)

使用するストアの型。

Alias (別名)

String (文字列)

キーストアに多数の非公開キーがある場合、使用する鍵の別名を示します。定義しない場合、ファイルにある最初のキーがデフォルトで使用されます。

Key Password (キーパスワード)

String (文字列)

非公開キーを保護するために使用するパスワード。

Password (パスワード)

String (文字列)

キーストアを保護するために使用するパスワード。

Algorithm (アルゴリズム)

String (文字列)

キーストアで使用するアルゴリズム。

### [](#standard-revocation-check)Standard Revocation Check (標準の取り消し確認)
項目 型 説明 デフォルト値 必須

Only End Entities (最後のエンティティのみ)

Boolean (ブール)

証明書チェーンの最後の要素のみを確認します。

Prefer Crls (CRL を優先)

Boolean (ブール)

OCSP 証明証明書を使用する前に、CRL 署名証明書を使用します。

No Fallback (代替なし)

Boolean (ブール)

第 2 の確認方法 ([Prefer Crls (CRL を優先)] で選択されていない方法) を使用しません。

Soft Fail (ソフトフェール)

Boolean (ブール)

取り消しサーバーに到達できない場合や取り消しサーバーが混雑している場合に検証エラーを回避します。

### [](#custom-ocsp-responder)Custom OCSP Responder (カスタム OCSP レスポンダー)
項目 型 説明 デフォルト値 必須

Url

String (文字列)

OCSP レスポンダーの URL。

Cert Alias (証明書の別名)

String (文字列)

OCSP 応答の署名証明書の別名 (存在する場合)。署名証明書がトラストストア内にある必要があります。

### [](#crl-file)CRL File (CRL ファイル)
項目 型 説明 デフォルト値 必須

Path (パス)

String (文字列)

CRL ファイルへのパス。

### [](#Reconnection)Reconnection (再接続)
項目 型 説明 デフォルト値 必須

Fails Deployment (デプロイメントに失敗)

Boolean (ブール)

アプリケーションがデプロイされると、すべてのコネクタで接続テストが実行されます。true に設定されている場合、各コネクタへのデプロイメント接続テストに失敗すると、デプロイメントが失敗します。

Reconnection Strategy (再接続戦略)

使用する再接続戦略。

### [](#reconnect)Reconnect (再接続)
項目 型 説明 デフォルト値 必須

Frequency (頻度)

Number (数値)

再接続する頻度 (ミリ秒)。

Count (数)

Number (数値)

再接続を試みる回数

blocking (ブロック)

Boolean (ブール)

false の場合、再接続戦略が個別の非ブロックスレッドで実行されます。

true

### [](#reconnect-forever)Reconnect Forever (繰り返し再接続)
項目 型 説明 デフォルト値 必須

Frequency (頻度)

Number (数値)

再接続する頻度 (ミリ秒)。

blocking (ブロック)

Boolean (ブール)

false の場合、再接続戦略が個別の非ブロックスレッドで実行されます。

true

### [](#ExpirationPolicy)Expiration Policy (有効期限ポリシー)
項目 型 説明 デフォルト値 必須

Max Idle Time (最大アイドル時間)

Number (数値)

有効期限の対象とみなされるまで、動的設定インスタンスがアイドル状態を維持できる最大時間のスカラー時間値

Time Unit (時間単位)

Enumeration (列挙)。次のいずれかになります。

  • NANOSECONDS (ナノ秒)

  • MICROSECONDS (マイクロ秒)

  • MILLISECONDS (ミリ秒)

  • SECONDS (秒)

  • MINUTES (分)

  • HOURS (時間)

  • DAYS (日)

maxIdleTime 属性の時間単位

### [](#AnypointMQMessageAttributes)Anypoint MQ メッセージの属性
項目 型 説明 デフォルト値 必須

Destination (宛先)

String (文字列)

メッセージの宛先。

x

Headers (ヘッダー)

Object (オブジェクト)

メッセージヘッダー。

x

Properties (プロパティ)

Object (オブジェクト)

メッセージプロパティ。

x

Message Id (メッセージ ID)

String (文字列)

メッセージの ID。

x

Delivery Count (配信数)

Number (数値)

メッセージが配信された回数。

x

Content Type (コンテンツタイプ)

String (文字列)

メッセージ本文のコンテンツタイプ。

Ack Token (Ack トークン)

String (文字列)

MANUAL モードでのみ、手動 ACK および NACK 操作を実行するために使用される肯定応答トークン。

### [](#circuitBreaker)サーキットブレーカー
項目 型 説明 デフォルト値 必須

On Error Types (エラー時種別)

String (文字列)

フローの実行中に発生したサーキットの障害と同等のエラー種別。デフォルトでは、すべてのエラーが障害とみなされます。

Errors Threshold (エラーしきい値)

Number (数値)

サーキットで考慮される、サーキットがオープンになるのに発生する必要があるエラーの回数。

x

Trip Timeout (トリップタイムアウト)

Number (数値)

errorsThreshold​ に達してからサーキットがオープンのままである期間。

x

Trip Timeout Unit (トリップタイムアウト単位)

Enumeration (列挙)。次のいずれかになります。

  • NANOSECONDS (ナノ秒)

  • MICROSECONDS (マイクロ秒)

  • MILLISECONDS (ミリ秒)

  • SECONDS (秒)

  • MINUTES (分)

  • HOURS (時間)

  • DAYS (日)

tripTimeout 値にバインドされる時間単位。

MILLISECONDS

### [](#repeatable-in-memory-stream)Repeatable In Memory Stream (反復可能なメモリ内ストリーム)
項目 型 説明 デフォルト値 必須

Initial Buffer Size (初期バッファサイズ)

Number (数値)

ストリームをコンシュームしてランダムにアクセスするために割り当てられるメモリ量。ストリームのデータ量がこのバッファサイズを超える場合は、bufferSizeIncrement 属性に従って、maxInMemorySize を上限としてバッファが拡張されます。

Buffer Size Increment (バッファサイズ増分)

Number (数値)

バッファサイズをどれだけ拡張できるか。0 以下に設定した場合、バッファがフルになると ​STREAM_MAXIMUM_SIZE_EXCEEDED​ が発生します。

Max Buffer Size (最大バッファサイズ)

Number (数値)

STREAM_MAXIMUM_SIZE_EXCEEDED​ が発生する前に使用できるメモリの最大量。0 以下に設定した場合、メモリは無制限になります。

Buffer Unit (バッファ単位)

Enumeration (列挙)。次のいずれかになります。

  • BYTE

  • KB

  • MB

  • GB

これらのすべての属性の単位

### [](#repeatable-file-store-stream)Repeatable File Store Stream (反復可能なファイルストアストリーム)
項目 型 説明 デフォルト値 必須

Max In Memory Size (最大メモリ内サイズ)

Number (数値)

コンテンツがディスクにバッファされる前にストリームがローカルに保持できるメモリの最大量。

Buffer Unit (バッファ単位)

Enumeration (列挙)。次のいずれかになります。

  • BYTE

  • KB

  • MB

  • GB

maxInMemorySize で表される単位。

### [](#RedeliveryPolicy)Redelivery Policy (再配信ポリシー) 再配信ポリシーについての詳細は、​[「再配信ポリシーの設定」](https://docs.mulesoft.com/mule-runtime/latest/redelivery-policy)​を参照してください。 ### [](#AnypointMQMessagePublishAttributes)Anypoint MQ Message Publish Attributes (Anypoint MQ メッセージのパブリッシュ属性)
項目 型 説明 デフォルト値 必須

Message Id (メッセージ ID)

String (文字列)

パブリッシュされたメッセージの ID。

x

### [](#prefetch)プリフェッチ
項目 型 説明 デフォルト値 必須

Max Local Messages (最大ローカルメッセージ数)

Number (数値)

サービスからプリフェッチされ、フローにディスパッチされる前にローカルに保持されるメッセージの最大数。

Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]

Unresolved include directive in modules/ROOT/pages/anypoint-mq-connector-reference.adoc - include::connectors::partial$mq-disclaimer.adoc[]

30

### [](#polling)ポーリング
項目 型 説明 デフォルト値 必須

Fetch size (フェッチサイズ)

Number (数値)

ポーリングの実行ごとにフェッチするメッセージの最大数 (1 ~ 10)。

10 よりも大きい値を設定すると、エラーが発生します。

10

Scheduling Strategy (スケジュール戦略)

scheduling-strategy

サービスからのメッセージのフェッチをトリガーするスケジュール戦略。

x

### [](#fixed-frequency)一定間隔
項目 型 説明 デフォルト値 必須

Frequency (頻度)

Number (数値)

Polling frequency (ポーリング頻度)

1000ms

(1 秒)

Start delay (開始遅延)

Number (数値)

開始前にスケジューラーが待機する時間。

10

Time unit (時間単位)

Enumeration (列挙)。次のいずれかになります。

  • DAYS

  • HOURS

  • MINUTES

  • SECONDS

  • MILLISECONDS

startDelay​ 設定で使用する時間単位。

MILLISECONDS

### [](#cron)Cron
項目 型 説明 デフォルト値 必須

Expression (式)

Cron 式

x

Time Zone (タイムゾーン)

String (文字列)

式が基づくタイムゾーンの ID。

## [](#関連情報)関連情報 - [一定間隔](https://docs.mulesoft.com/mule-runtime/latest/scheduler-concept#fixed-frequency) - [Cron 式](https://docs.mulesoft.com/mule-runtime/latest/scheduler-concept#cron-expressions) # Anypoint MQ Consume 操作 Anypoint MQ Connector の ​`consume`​ 操作を使用すると、アプリケーションはユースケースをサポートする肯定応答戦略を使用して、フローの実行中にいつでも任意の宛先から単一メッセージをコンシュームできます。 ## [](#受信メッセージ構造)受信メッセージ構造 すべての受信メッセージは 2 つの部分から構成されます。 - `payload`​ - メッセージ本文を表します。 - `attributes`​ - 受信したメッセージのメタデータ (ID や他のプロパティなど) が含まれます。 ![Anypoint MQ メッセージの属性](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-consume-attributes.png) ## [](#メッセージのコンシューム)メッセージのコンシューム キューからメッセージをコンシュームする最も簡単な方法は、Mule パレットからキャンバスに ​**\[Consume\]**​ 操作をドラッグして、メッセージの取得元であるキューの名前で ​`Queue`​ パラメーターを設定することです。 ![Anypoint MQ Consume 操作](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-consume-operation.png) ```xml ``` この例の Consume 操作は、​`myQueue`​ で最初に使用可能なメッセージをコンシュームし、次の構造の ​`MuleMessage`​ インスタンスに変換します。 - ペイロードとしてのメッセージ本文。 - メッセージ属性に含まれるメッセージのメタデータ。属性には ​`messageId`​ または ​`redeliveryCount`​ が含まれます。 `Queue`​ パラメーターで DataWeave 式を使用して、メッセージのコンシューム元となる対象キューを動的に変更します。 ## [](#コンシュームされるメッセージ肯定応答)コンシュームされるメッセージ肯定応答 - **自動** デフォルトでは、​`acknowledgementMode`​ は ​`IMMEDIATE`​ です。つまり、メッセージはキューからコンシュームされるとすぐに、Mule フローに配信される前に肯定応答されます。メッセージの肯定応答に失敗すると、Consume 操作は ​`ANYPOINT-MQ:ACKING`​ エラーで終了します。 詳細は、​[「即時肯定応答」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack#immediate-acknowledgment)​を参照してください。 - **手動** メッセージの肯定応答を手動で制御するには、​`acknowledgementMode`​ を ​`MANUAL`​ に変更します。​`MANUAL`​ 肯定応答モードを使用している場合、アプリケーションロジックは ​[ACK](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack)​ または ​[NACK](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack)​ 操作を使用して、メッセージの肯定応答を実行するタイミングを決定します。 手動肯定応答を実行するには、結果のメッセージ属性の一部として提供される ​`ackToken`​ の値が必要です。 詳細は、​[「手動肯定応答」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack#manual-acknowledgment)​を参照してください。 肯定応答タイムアウトについての詳細は、​[「肯定応答タイムアウト」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack#acknowledgment-timeout)​を参照してください。 ## [](#空のキュー管理)空のキュー管理 キューにメッセージがある場合、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 - ヘッダーマップ - プロパティマップ ![Anypoint MQ メッセージの属性](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-consume-attributes.png) `attributes`​ 構造についての詳細は、​[「Anypoint MQ Connector リファレンス - 3.x」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-connector-reference#consume)​の Consume 操作を参照してください。 ## [](#関連情報)関連情報 - [Anypoint MQ ACK および NACK 操作](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack) # Anypoint MQ Consume 操作 Anypoint MQ Connector の Consume 操作を使用すると、アプリケーションはユースケースをサポートする肯定応答戦略を使用して、フローの実行中にいつでもキューから単一メッセージをコンシュームできます。

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

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

## [](#受信メッセージ構造)受信メッセージ構造 すべての受信メッセージは 2 つの部分から構成されます。 - `payload`​ - メッセージ本文を表します。 - `attributes`​ - 受信したメッセージのメタデータ (ID や他のプロパティなど) が含まれます。 ![ペイロードとメッセージ属性を含む Mule メッセージ](/mulesoft/current/ja-jp/anypoint-mq-connector/latest/_images/amq-4x-consume-attributes.png) ## [](#メッセージのコンシューム)メッセージのコンシューム キューからメッセージをコンシュームする最も簡単な方法は、Mule パレットからキャンバスに ​**\[Consume\]**​ 操作をドラッグして、メッセージの取得元であるキューの名前で ​`Queue`​ パラメーターを設定することです。 ![宛先、肯定応答モードとタイムアウト、ポーリングタイムアウトを含むコネクタ設定](/mulesoft/current/ja-jp/anypoint-mq-connector/latest/_images/amq-4x-consume-operation.png) ```xml ``` この例の Consume 操作は、​`myQueue`​ で最初に使用可能なメッセージをコンシュームし、次の構造の ​`MuleMessage`​ インスタンスに変換します。 - ペイロードとしてのメッセージ本文。 - メッセージ属性に含まれるメッセージのメタデータ。属性には ​`messageId`​ または ​`redeliveryCount`​ が含まれます。 `Queue`​ パラメーターで DataWeave 式を使用して、メッセージのコンシューム元となる対象キューを動的に変更します。 ## [](#コンシュームされるメッセージ肯定応答)コンシュームされるメッセージ肯定応答 - **自動** デフォルトでは、​`acknowledgementMode`​ は ​`IMMEDIATE`​ です。つまり、メッセージはキューからコンシュームされるとすぐに、Mule フローに配信される前に肯定応答されます。メッセージの肯定応答に失敗すると、Consume 操作は ​`ANYPOINT-MQ:ACKING`​ エラーで終了します。 詳細は、​[「即時肯定応答」](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack#immediate-acknowledgment)​を参照してください。 - **手動** メッセージの肯定応答を手動で制御するには、​`acknowledgementMode`​ を ​`MANUAL`​ に変更します。​`MANUAL`​ 肯定応答モードを使用している場合、アプリケーションロジックは ​[ACK](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack)​ または ​[NACK](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack)​ 操作を使用して、メッセージの肯定応答を実行するタイミングを決定します。 手動肯定応答を実行するには、結果のメッセージ属性の一部として提供される ​`ackToken`​ の値が必要です。 詳細は、​[「手動肯定応答」](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack#manual-acknowledgment)​を参照してください。 肯定応答タイムアウトについての詳細は、​[「肯定応答タイムアウト」](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack#acknowledgment-timeout)​を参照してください。 ## [](#空のキュー管理)空のキュー管理 キューにメッセージがある場合、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 メッセージ](/mulesoft/current/ja-jp/anypoint-mq-connector/latest/_images/amq-4x-consume-attributes.png) `attributes`​ 構造についての詳細は、​[「Anypoint MQ Connector リファレンス - 4.x」](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-connector-reference#consume)​の Consume 操作を参照してください。 ## [](#関連情報)関連情報 - [Anypoint MQ ACK および NACK 操作](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack) # Anypoint MQ Publish 操作 Publish 操作では、新しい Anypoint MQ メッセージを作成し、指定した宛先 (キュー、FIFO キュー、メッセージエクスチェンジ) に送信できます。この操作を使用して、メッセージのコンテンツと、必要なすべてのヘッダーとプロパティの両方を設定できます。 ## [](#送信メッセージ構造)送信メッセージ構造 メッセージには次のパラメーターが含まれます。 - `body` パブリッシュされた Anypoint MQ メッセージの本文が含まれます。 - `properties` メッセージのプロパティである文字列のシンプルなキー - 値のマップを表します。 ## [](#メッセージの送信)メッセージの送信 デフォルトの形式で使用された場合、コネクタによって受信メッセージの ​`payload`​ で見つかったコンテンツが文字列としてシリアル化された Anypoint MQ メッセージの ​`body`​ として、宣言された ​`destination`​ にパブリッシュされます。 ![Anypoint MQ Publish 操作](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-publish-operation.png) ```xml ``` この操作では、次の出力が生成されます。 - ペイロードとして送信された本文のコピー たとえば、「Hello world」メッセージを送信する場合、結果のペイロードは「Hello world」文字列になります。 - メッセージ属性に含まれるメッセージの ​`messageId`​ 値 ![Anypoint MQ Publish 操作の出力属性](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-publish-attributes.png) ## [](#ユーザープロパティの設定)ユーザープロパティの設定 たとえば、他のメッセージングシステムとの互換性の提供、またはメッセージのコンテンツタイプの通信などのために、送信メッセージのプロパティを定義できます。 Anypoint MQ は、プロパティ値で文字列と数値をサポートしています。
数値プロパティ値に基づくルーティングルールを作成し、Anypoint MQ Connector を使用してメッセージをパブリッシュする場合、Anypoint MQ Connector バージョン 3.2.0 以降にアップグレードする必要があります。 これより前のバージョンのコネクタでは、すべてのプロパティ値が文字列として送信されるため、メッセージは数値プロパティに基づくルーティングルールに一致しません。
これらのプロパティをメッセージで直接設定するには、Publish 操作の ​`properties`​ パラメーターを使用し、DataWeave を使用してキーと値のマップを定義します。 次に例を示します。 ![プロパティを使用する Anypoint MQ Publish](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-publish-properties.png) ```xml ``` Publish 操作で定義したユーザープロパティは、Anypoint MQ のメッセージの ​**\[Message Browser (メッセージブラウザー)\]**​ 詳細ペインに表示されます。 ![「Message Browser (メッセージブラウザー)」 詳細ペインの 「User Properties (ユーザープロパティ)」](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-publish-user-props.png) Figure 1. この画面では、(​**1**​) ​**\[Message Browser (メッセージブラウザー)\]**​ オプションとその詳細ペインに (​**2**​) ​**\[User Properties (ユーザープロパティ)\]**​ が表示されています。 ### [](#メッセージ本文の変換)メッセージ本文の変換 ペイロードの形式が正しくない場合、正しい形式に変換できます。 ペイロードを変換するには、Publish 操作の ​`body`​ パラメーターでインライン DataWeave 変換を宣言します。 ![変換を使用する Anypoint MQ Publish](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-publish-transformed.png) ```xml ``` このケースでは、操作の結果は JSON として​_変換された_​値になります。 ## [](#詳細な設定)詳細な設定 Anypoint MQ で一意のメッセージ ID を生成できるようにする代わりに、一意のカスタムメッセージ ID を設定できます。プロパティとしてメッセージ本文のコンテンツタイプを追加することもできます。 ### [](#カスタムメッセージ-id-の設定)カスタムメッセージ ID の設定 Anypoint MQ で新しくパブリッシュされたメッセージの一意の ​`messageId`​ 値を生成できるようにする代わりに、​**\[Advanced (詳細)\]**​ タブの ​**\[Message Id (メッセージ ID)\]**​ 項目でカスタム ​`messageId`​ 値を設定します。 ![Anypoint MQ Publish の \[Advanced (詳細)\] タブ - \[Message Id (メッセージ ID)\]](_images/amq-3x-publish-advanced.png) Anypoint MQ メッセージの ID は一意である必要があります。重複する ID の不要な副作用を避けるために一意のカスタム ID を選択します。 FIFO キューでは、メッセージの重複を排除できます。同じメッセージ ID が含まれる FIFO キューに複数のメッセージを送信した場合、最初に到着したメッセージが有効なメッセージとして保持されます。 詳細は、​[「FIFO の正確に 1 回の配信」](https://docs.mulesoft.com/jp/mq/mq-queues#fifo-exactly-once-delivery)​を参照してください。 ### [](#メッセージ配信の遅延)メッセージ配信の遅延 Anypoint MQ では、メッセージをパブリッシュし、​`deliveryDelay`​ パラメーターを使用してコンシューマーに表示されるまで延期する機能がサポートされています。 1. Publish 操作を選択します。 2. **\[Advanced (詳細)\]**​ タブをクリックします。 3. **\[Delivery Delay (配信遅延)\]**​ パラメーターを遅延する時間に設定します。 ![Anypoint MQ Publish の \[Advanced (詳細)\] タブ - \[Delivery Delay (配信遅延)\]](_images/amq-3x-publish-advanced-delay.png) デフォルトでは、​`deliveryDelay`​ パラメーターは ​`0`​ に設定されています。これはメッセージをすぐに (遅延なしで) 配信することを示します。​`deliveryDelay`​ の最大値は 15 分です。 詳細は、​[「遅延したメッセージの配信」](https://docs.mulesoft.com/jp/mq/mq-glossary#delay-messages)​を参照してください。 4. **\[Delivery Delay Unit (配信遅延単位)\]**​ メニューから単位を選択します。 ### [](#メッセージグループ-id-の設定)メッセージグループ ID の設定 FIFO キューを使用する場合、​`messageGroupID`​ パラメーターを使用して送信メッセージにメッセージグループ ID を割り当てることができます。 1. Publish 操作を選択します。 2. **\[Advanced (詳細)\]**​ タブをクリックします。 3. **\[Message Group ID (メッセージグループ ID)\]**​ 項目に ID を入力します。 ![Anypoint MQ Publish の \[Advanced (詳細)\] タブ - \[Message Group ID (メッセージグループ ID)\]](_images/amq-3x-publish-advanced-groupid.png) デフォルトでは、FIFO キューに送信されるすべてのメッセージに自動生成された同じグループ ID が割り当てられます。 詳細は、​[「FIFO キューとメッセージグループ」](https://docs.mulesoft.com/jp/mq/mq-queues#fifo-queues-msg-groups)​を参照してください。 メッセージグループを最大限に活用するには、各グループのメッセージ数を少なくしてより詳細なグループを定義します。 たとえば、複数のユーザーからのメッセージを処理する場合、各ユーザーのイベントの相対的な順序を維持するために、ユーザーの ​`userId`​ 値をメッセージグループ ID として使用します。 ### [](#コンテンツタイプの伝播)コンテンツタイプの伝播 メッセージ本文のコンテンツタイプが Anypoint MQ メッセージのプロパティとして伝播されるかどうかを指定するには、​`sendContentType`​ パラメーターを使用します。 このパラメーターを ​`true`​ に設定すると、自動的に ​`contentType`​ プロパティが送信メッセージに追加されます。これは、メッセージ本文が ​`application/json`​ または ​`application/xml`​ 形式であることを宣言する場合に役立ちます。 ## [](#関連情報)関連情報 - [Anypoint MQ Subscriber ソース](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-listener) - [Anypoint MQ の用語集](https://docs.mulesoft.com/jp/mq/mq-glossary) - [キューの設定と使用](https://docs.mulesoft.com/jp/mq/mq-queues) # Anypoint MQ Publish 操作 Publish 操作では、新しい Anypoint MQ メッセージを作成し、指定した宛先 (キュー、FIFO キュー、メッセージエクスチェンジ) に送信できます。この操作を使用して、メッセージのコンテンツと、必要なすべてのヘッダーとプロパティの両方を設定できます。 ## [](#送信メッセージ構造)送信メッセージ構造 メッセージには次のパラメーターが含まれます。 - `body` パブリッシュされた Anypoint MQ メッセージの本文が含まれます。 - `properties` メッセージのプロパティである文字列のシンプルなキー - 値のマップを表します。 ## [](#メッセージの送信)メッセージの送信 デフォルトの形式で使用された場合、コネクタによって受信メッセージの ​`payload`​ で見つかったコンテンツが文字列としてシリアル化された Anypoint MQ メッセージの ​`body`​ として、宣言された ​`destination`​ にパブリッシュされます。 ![\[Destination (宛先)\] が \[invoiceQueue\] に設定されている Publish 設定ウィンドウ](_images/amq-4x-publish-operation.png) ```xml ``` この操作では、次の出力が生成されます。 - ペイロードとして送信された本文のコピー たとえば、「Hello world」メッセージを送信する場合、結果のペイロードは「Hello world」文字列になります。 - メッセージ属性に含まれるメッセージの ​`messageId`​ 値 ![ペイロードとメッセージ属性を含む Mule メッセージ出力](/mulesoft/current/ja-jp/anypoint-mq-connector/latest/_images/amq-4x-publish-attributes.png) ## [](#ユーザープロパティの設定)ユーザープロパティの設定 たとえば、他のメッセージングシステムとの互換性の提供、またはメッセージのコンテンツタイプの通信などのために、送信メッセージのプロパティを定義できます。 Anypoint MQ は、プロパティ値で文字列と数値をサポートしています。
数値プロパティ値に基づくルーティングルールを作成し、Anypoint MQ Connector を使用してメッセージをパブリッシュする場合、Anypoint MQ Connector バージョン 3.2.0 以降にアップグレードする必要があります。 これより前のバージョンのコネクタでは、すべてのプロパティ値が文字列として送信されるため、メッセージは数値プロパティに基づくルーティングルールに一致しません。
これらのプロパティをメッセージで直接設定するには、Publish 操作の ​`properties`​ パラメーターを使用し、DataWeave を使用してキーと値のマップを定義します。 次に例を示します。 ![AUTH\_TYPE 属性と AUTH\_TOKEN 属性が表示されているコネクタ設定](/mulesoft/current/ja-jp/anypoint-mq-connector/latest/_images/amq-4x-publish-properties.png) ```xml ``` Publish 操作で定義したユーザープロパティは、Anypoint MQ のメッセージの ​**\[Message Browser (メッセージブラウザー)\]**​ 詳細ペインに表示されます。 ![メッセージのユーザープロパティが表示されている \[Message Browser (メッセージブラウザー)\]](_images/amq-4x-publish-user-props.png) ### [](#メッセージ本文の変換)メッセージ本文の変換 ペイロードの形式が正しくない場合、正しい形式に変換できます。 ペイロードを変換するには、Publish 操作の ​`body`​ パラメーターでインライン DataWeave 変換を宣言します。 ![DataWeave 変換が表示されているコネクタ設定](/mulesoft/current/ja-jp/anypoint-mq-connector/latest/_images/amq-4x-publish-transformed.png) ```xml ``` このケースでは、操作の結果は JSON として​_変換された_​値になります。 ## [](#詳細な設定)詳細な設定 Anypoint MQ で一意のメッセージ ID を生成できるようにする代わりに、一意のカスタムメッセージ ID を設定できます。プロパティとしてメッセージ本文のコンテンツタイプを追加することもできます。 ### [](#カスタムメッセージ-id-の設定)カスタムメッセージ ID の設定 Anypoint MQ で新しくパブリッシュされたメッセージの一意の ​`messageId`​ 値を生成できるようにする代わりに、​**\[Advanced (詳細)\]**​ タブの ​**\[Message Id (メッセージ ID)\]**​ 項目でカスタム ​`messageId`​ 値を設定します。 ![\[Message ID (メッセージ ID)\] が \[12345\] に設定されている Publish 設定ウィンドウの \[Advanced (詳細)\] タブ](_images/amq-4x-publish-advanced.png) Anypoint MQ メッセージの ID は一意である必要があります。重複する ID の不要な副作用を避けるために一意のカスタム ID を選択します。 FIFO キューでは、メッセージの重複を排除できます。同じメッセージ ID が含まれる FIFO キューに複数のメッセージを送信した場合、最初に到着したメッセージが有効なメッセージとして保持されます。 詳細は、​[「FIFO の正確に 1 回の配信」](https://docs.mulesoft.com/jp/mq/mq-queues#fifo-exactly-once-delivery)​を参照してください。 ### [](#メッセージ配信の遅延)メッセージ配信の遅延 Anypoint MQ では、メッセージをパブリッシュし、​`deliveryDelay`​ パラメーターを使用してコンシューマーに表示されるまで延期する機能がサポートされています。 1. Publish 操作を選択します。 2. **\[Advanced (詳細)\]**​ タブをクリックします。 3. **\[Delivery Delay (配信遅延)\]**​ パラメーターを遅延する時間に設定します。 ![\[Delivery Delay (配信遅延)\] が \[2000\] に設定されている Publish 設定ウィンドウの \[Advanced (詳細)\] タブ](_images/amq-4x-publish-advanced-delay.png) デフォルトでは、​`deliveryDelay`​ パラメーターは ​`0`​ に設定されています。これはメッセージをすぐに (遅延なしで) 配信することを示します。​`deliveryDelay`​ の最大値は 15 分です。 詳細は、​[「遅延したメッセージの配信」](https://docs.mulesoft.com/jp/mq/mq-glossary#delay-messages)​を参照してください。 4. **\[Delivery Delay Unit (配信遅延単位)\]**​ メニューから単位を選択します。 ### [](#メッセージグループ-id-の設定)メッセージグループ ID の設定 FIFO キューを使用する場合、​`messageGroupID`​ パラメーターを使用して送信メッセージにメッセージグループ ID を割り当てることができます。 1. Publish 操作を選択します。 2. **\[Advanced (詳細)\]**​ タブをクリックします。 3. **\[Message Group ID (メッセージグループ ID)\]**​ 項目に ID を入力します。 ![\[Message Group ID (メッセージグループ ID)\] が \[#\[vars.userId\]\] に設定されている Publish 設定ウィンドウの \[Advanced (詳細)\] タブ](_images/amq-4x-publish-advanced-groupid.png) デフォルトでは、FIFO キューに送信されるすべてのメッセージに自動生成された同じグループ ID が割り当てられます。 詳細は、​[「FIFO キューとメッセージグループ」](https://docs.mulesoft.com/jp/mq/mq-queues#fifo-queues-msg-groups)​を参照してください。 メッセージグループを最大限に活用するには、各グループのメッセージ数を少なくしてより詳細なグループを定義します。 たとえば、複数のユーザーからのメッセージを処理する場合、各ユーザーのイベントの相対的な順序を維持するために、ユーザーの ​`userId`​ 値をメッセージグループ ID として使用します。 ### [](#コンテンツタイプの伝播)コンテンツタイプの伝播 メッセージ本文のコンテンツタイプが Anypoint MQ メッセージのプロパティとして伝播されるかどうかを指定するには、​`sendContentType`​ パラメーターを使用します。 このパラメーターを ​`true`​ に設定すると、自動的に ​`contentType`​ プロパティが送信メッセージに追加されます。これは、メッセージ本文が ​`application/json`​ または ​`application/xml`​ 形式であることを宣言する場合に役立ちます。 ## [](#関連情報)関連情報 - [Anypoint MQ Subscriber ソース](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-listener) - [Anypoint MQ の用語集](https://docs.mulesoft.com/jp/mq/mq-glossary) - [キューの設定と使用](https://docs.mulesoft.com/jp/mq/mq-queues) # Anypoint MQ Subscriber ソース Anypoint MQ Connector の Subscriber ソースは、アプリケーションで新しいメッセージをリスンし、宛先に到達したメッセージをコンシュームできます。さまざまなリスニング戦略を設定し、パフォーマンス、予測可能性、スケジュールに合わせてコンシュームを調整できます。 ## [](#最大メッセージスループット)最大メッセージスループット デフォルトでは、Subscriber ソースは継続的リスニングモードで機能し、キューに到達するとすぐにメッセージをコンシュームしてスループットを最大化します。これは、​`prefetch`​ subscriber-type 設定と呼ばれます。​`prefetch`​ モードを使用する場合、Subscriber ソースはメッセージのローカルバッファが常にフルになるように試みます。これにより、アプリケーションでメッセージを受け入れ可能になり次第、Mule フローにディスパッチできます。 ![プリフェッチサブスクライバー種別](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-subscriber-prefetch.png) Figure 1. 矢印は、​**\[Subscriber (サブスクライバー)\]**​ プロパティウィンドウで ​**\[Prefetch (プリフェッチ)\]**​ に設定された ​**\[Subscriber type (サブスクライバー種別)\]**​ を示しています。 Unresolved include directive in modules/ROOT/pages/anypoint-mq-listener.adoc - include::connectors::partial$mq-disclaimer.adoc\[\] `prefetch`​ subscriber-type 設定で、​`maxLocalMessages`​ パラメーターを設定してフルバッファの対象サイズを指定し、どれだけ迅速にキューからメッセージが取得されるのかを制御します。 ```xml ``` - より大きなバッファサイズを使用してパフォーマンスを最大化する。他のコンシューマーはローカルで保持されるメッセージを使用できないことに留意してください。 コンシューマーがキューからメッセージを取得してローカルバッファに保存すると、メッセージはインフライトメッセージとして他のコンシューマーに表示され、他のコンシューマーはコンシュームできなくなります。メッセージは、アプリケーションで処理を開始するまで必要に応じてこのインフライト状態のままになります。 - より小さなバッファサイズを使用して、キューにパブリッシュされたときにメッセージを処理する。 バッファを小さくすると、アプリケーションレベルのスループットが制限され、競合するコンシューマーがブロックされることを回避できます。 Unresolved include directive in modules/ROOT/pages/anypoint-mq-listener.adoc - include::connectors::partial$mq-disclaimer.adoc\[\] ### [](#fifo-and-prefetch)FIFO キューとプリフェッチモード `prefetch`​ モードを使用する場合、​`acknowledgementTimeout`​ タイマーは、メッセージがキューから取得されたときに開始されます。 Anypoint MQ は、​`acknowledgementTimeout`​ で指定された有効期限 (時間) を順守しようとします。 ただし、メッセージがバッファに保持されている時間がタイムアウト値の 80% に達すると、Anypoint MQ は有効期限タイムアウトを延長します。 Unresolved include directive in modules/ROOT/pages/anypoint-mq-listener.adoc - include::connectors::partial$mq-disclaimer.adoc\[\] ## [](#予測可能なメッセージのコンシューム)予測可能なメッセージのコンシューム アプリケーションで最大スループットではなく、メッセージのコンシュームを予測して制御する必要がある場合、ポーリングソースとして Subscriber を設定できます。Subscriber ソースは、固定スケジュールレートでキューの新規メッセージをチェックします。ポーリングがトリガーされるたびに、Subscriber ソースは 1 ~ 10 個のメッセージを取得して、フローに 1 つずつ順番にディスパッチします。 メッセージは、すぐにコンシュームされません。代わりに、Subscriber ソースの次のポーリングがトリガーされるまでキューに保持されます。キューには、取得されるメッセージが最大 10 個含まれます。たとえば、キューに 15 個のメッセージがある場合、1 つのポーリングで最初の 10 個のメッセージがコンシュームされ、残りの 5 個のメッセージは次のポーリングを待機します。 Subscriber をポーリングソースとして設定するには、​`polling`​ subscriber-type を指定します。 デフォルトの ​`polling`​ subscriber-type の宣言は次のようになります。 ![ポーリングサブスクライバー種別](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-subscriber-polling.png) Figure 2. 矢印は、​**\[Subscriber (サブスクライバー)\]**​ プロパティウィンドウで ​**\[Polling (ポーリング)\]**​ に設定された ​**\[Subscriber type (サブスクライバー種別)\]**​ を示しています。 ```xml ``` デフォルトの ​`polling`​ subscriber-type の宣言では、宣言された宛先から 1 秒ごとにメッセージ 10 個の取得を試み、取得したメッセージを ​`MuleMessage`​ インスタンスとして 1 つずつフローにディスパッチします。 結果の ​`MuleMessage`​ には、以下が含まれます。 - メッセージのペイロードとしての本文 - メッセージ属性に含まれるメッセージのメタデータ ![メッセージ出力のペイロードと属性](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-subscriber-output.png) Figure 3. このスクリーンショットは、​**\[Output (出力)\]**​ タブのメッセージの (​**1**​) ​**\[Payload (ペイロード)\]**​ と (​**2**​) ​**\[Attributes (属性)\]**​ を示しています。 標準 Mule Runtime Engine スケジュール戦略を使用して、Subscriber ソースのポーリング戦略をカスタマイズできます。 - [一定間隔](https://docs.mulesoft.com/mule-runtime/latest/scheduler-concept#fixed-frequency) - [Cron 式](https://docs.mulesoft.com/mule-runtime/latest/scheduler-concept#cron-expressions) ### [](#カスタマイズされた固定ポーリング)カスタマイズされた固定ポーリング Subscriber ソースの新規メッセージのポーリング頻度はカスタマイズできます。これを行うには、カスタマイズされた ​`fixed-frequency`​ スケジュール戦略を使用して、​`polling`​ subscriber-type 戦略を宣言します。 デフォルトの ​`fixed-frequency`​ スケジュール戦略宣言は次のようになります。 ![一定間隔のスケジュール戦略](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-subscriber-polling-fixed.png) Figure 4. 矢印は、​**\[Subscriber (サブスクライバー)\]**​ プロパティウィンドウで ​**\[Fixed Frequency (一定間隔)\]**​ に設定された ​**\[Scheduling Strategy (スケジュール戦略)\]**​ を示しています。 ```xml ``` ### [](#cron-ベースのポーリング)cron ベースのポーリング Anypoint MQ サブスクライバーは、​`cron`​ スケジュール戦略を使用することもできます。このスケジュール戦略を使用すると、「毎日午後 2 時から 2 時 59 分まで、毎分実行」のようなジョブのスケジュールが可能になります。このスケジュール戦略を使用するには、カスタマイズされた ​`cron`​ スケジュール戦略を使用して ​`polling`​ subscriber-type 戦略を宣言します。 ![Cron スケジュール戦略](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-subscriber-polling-cron.png) Figure 5. 矢印は、​**\[Subscriber (サブスクライバー)\]**​ プロパティウィンドウで ​**\[Cron\]**​ に設定された ​**\[Scheduling Strategy (スケジュール戦略)\]**​ を示しています。 ```xml ``` ## [](#メッセージ肯定応答)メッセージ肯定応答 フローの実行が正常に完了した場合や、処理前にアプリケーションがメッセージをコンシュームした場合にのみ受信したメッセージに自動的に肯定応答できます。 - **自動** デフォルトでは、Subscriber ソースは自動肯定応答モードを使用します。このモードでは、メッセージフローの処理が成功した後で Subscriber ソースが取得するメッセージが自動的に肯定応答されます。つまり、Subscriber ソースはメッセージを取得してフローにディスパッチし、どのようにメッセージの処理が完了するのかを確認するために待機します。例外なしで処理が完了した場合にのみ ACK (肯定応答) を実行します。 例外が伝播されて処理フローの実行が完了すると、メッセージは自動的に肯定応答されず、再配信のためにキューに返されます。 詳細は、​[「自動肯定応答」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack#automatic-acknowledgment)​を参照してください。 - **即時** `IMMEDIATE`​ 肯定応答モードを使用する場合、コンシュームされたメッセージが肯定応答 (キューから削除) され、処理のために Mule フローにディスパッチされます。 メッセージの肯定応答に失敗すると、メッセージは破棄されます。メッセージはフローにディスパッチされず、肯定応答がタイムアウトするまでインフライトのままになります。 Unresolved include directive in modules/ROOT/pages/anypoint-mq-listener.adoc - include::connectors::partial$mq-disclaimer.adoc\[\] `IMMEDIATE`​ 肯定応答モードでプリフェッチモードの Subscriber ソースを使用すると、スレッドが累積したり、アプリケーションが無応答になったりする場合があります。 詳細は、​[「即時肯定応答」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack#immediate-acknowledgment)​を参照してください。 - **手動** 手動肯定応答モードを使用する場合、​[ACK](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack)​ または ​[NACK](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack)​ ソースを使用して、メッセージの肯定応答を実行するタイミングをアプリケーションロジックで決定します。 手動肯定応答を実行するには、結果のメッセージ属性の一部として提供される ​`ackToken`​ の値が必要です。 詳細は、​[「手動肯定応答」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack#manual-acknowledgment)​を参照してください。 肯定応答タイムアウトについての詳細は、​[「肯定応答タイムアウト」](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack#acknowledgment-timeout)​を参照してください。 ## [](#サーキットブレーカー機能)サーキットブレーカー機能 Subscriber ソースは、サーキットブレーク機能を提供します。この機能を使用すると、コンシュームされたメッセージの処理中に発生するエラーをコネクタでどのように処理するのかを制御できます。 たとえば、外部サービスに接続している場合、サーキットブレーカーを使用してそのサービスのダウンタイムを処理できます。サーキットブレーカーでは、要求の実行を停止したり、負荷を減らして外部サービスを回復したりできます。 Mule 4 アプリケーションでの Anypoint MQ の使用は、キューからメッセージをコンシュームし、外部サービスを使用して処理を行う、MQ サブスクライバーを使用する Mule フローがあることを意味します。このサービスが使用できない場合、次のようになります。 1. 要求が失敗する。 2. エラーが発生する。 3. メッセージの処理が失敗に終わるか、デッドレターキュー (DLQ) にメッセージを送信するなど、メッセージ処理のカスタムエラーで終了します。 外部サービスを利用できない場合、メッセージを処理するすべての試みは失敗し、アプリケーションは成功しないメッセージのコンシュームを繰り返し実行し続けます。この動作は、一定期間メッセージをさらにコンシュームしないように、サブスクライバーにエラーを通知すれば回避できます。 ### [](#サーキットブレーカープロセス)サーキットブレーカープロセス Subscriber ソースによって提供されるサーキットブレーカー機能は、Mule で提供されるエラー処理メカニズムにバインドされています。エラー通知メカニズムを使用してサーキット障害と呼ばれる外部サービスに関連するエラー数を記録します。サーキット障害にはあらゆるエラーをバインドできます。たとえば、​`HTTP:TIMEOUT`​ や ​`FTP:SERVICE_NOT_AVAILABLE`​、さらには ​`ORG:EXTERNAL_ERROR`​ のようなアプリケーションのカスタムエラーもバインドできます。 Mule フローの実行がエラーに終わった場合、Subscriber ソースはエラーが外部サービスエラーを示す ​`onErrorTypes`​ のいずれかであるかどうかをチェックし、​`errorsThreshold`​ に達するまで、連続発生数を数えます。 `errorsThreshold`​ に達すると、サーキットがトリップし、​`tripTimeout`​ によって指定される期間中、オープン状態のままになります。 サーキットがオープン状態の間、ポーリングまたはコンシュームされたすべてのメッセージは肯定応答 (NACK) されず、Anypoint MQ は ​`tripTimeout`​ が経過するまでその後の操作 (ポーリング、ACK、NACK、パブリッシュ) を停止します。 `tripTimeout`​ が経過すると、サーキットブレーカーは新しい​[状態](#circuit-breaker-states)​に移行します。 新しい状態が ​**\[Closed (クローズ)\]**​ の場合、クライアントは次のポーリングでメッセージをコンシュームできます。 アプリケーションを CloudHub にデプロイするワーカーが 1 つか複数かによって、この動作は異なります。 1 つのワーカー ワーカーはメッセージをキューから個別に処理し、エラー数が ​`errorsThreshold`​ の値に達するとサーキットブレーカーをオープン状態にします。 複数のワーカー 各ワーカーに関連付けられたサーキットブレーカーがあり、​`errorsThreshold`​ と ​`tripTimeout`​ の値を個別に追跡します。 各ワーカーはメッセージをキューから個別に処理し、エラー数がそのワーカーで設定された ​`errorsThreshold`​ の値に達するとサーキットブレーカーをオープン状態にします。 デフォルトでは、サーキットブレーク機能は無効になっています。 ### [](#circuit-breaker-states)サーキットブレーカーの状態 サーキットブレーカーは、3 つの状態 (クローズド、オープン、ハーフオープン) があります。アプリケーションの動作は、現在の状態に基づいて変わります。詳細は、 [「Microsoft Circuit Breaker pattern (Microsoft サーキットブレーカーパターン)」](https://docs.microsoft.com/en-us/azure/architecture/patterns/circuit-breaker)​を参照してください。 ![サーキットブレーカーの状態図](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-subscriber-states-view.png) - **クローズド** 通常 Subscriber ソースが設定に基づいて MQ からメッセージを取得する開始状態。実質的にサーキットブレーカーが存在しないかのように機能します。 - **クローズドからオープンへの遷移** メッセージの処理中に成功することなく失敗が連続して発生し、​`errorsThreshold`​ 値に達すると、サーキットブレーカーがトリップし、オープン状態に遷移します。 フローにすでにディスパッチされたメッセージは、結果が成功か失敗かに関係なく処理を完了します。 ブローカーのインフライトであるが、まだディスパッチされておらず、ローカルに保持されているメッセージは、肯定応答されず、別のコンシューマーに再配信するためにキューに戻されます。 - **オープン** Subscriber ソースはメッセージの取得を試みず、​`tripTimeout`​ に達するまで通知せずにイテレーションをスキップします。 - **ハーフオープン** `tripTimeout`​ の経過後、Subscriber ソースはハーフオープン状態になります。次回のメッセージのポーリングで、Subscriber ソースは通常のクローズド状態に戻る前にサービスからメッセージを 1 つ取得し、そのメッセージを使用してシステムが回復したかどうかをチェックします。 Subscriber ソースが 1 つのメッセージを正常にフェッチしてフローにディスパッチし、処理が正常に完了すると、Subscriber ソースは通常の状態に戻り、すぐに追加のメッセージのフェッチを試みます。 Mule フローの処理が、予期される ​`onErrorTypes`​ のいずれかで失敗すると、サーキットはオープン状態に戻り、​`tripTimeout`​ タイマーがリセットされます。 ### [](#サーキットブレーカーの設定)サーキットブレーカーの設定 サーキットブレーカーは、グローバルサーキットブレーカーまたは非公開サーキットブレーカーとして設定できます。 どちらの方法でも、設定パラメーターは同じです。 - `onErrorTypes` フローの実行中に障害とみなされるエラー種別。エラーの発生は、フローがエラーの伝達によって完了した場合にのみ計数されます。デフォルトでは、すべてのエラーがサーキット障害とみなされます。 - `errorsThreshold` サーキットブレーカーがオープンになるのに発生する必要がある ​`onErrorTypes`​ エラーの回数。 - `tripTimeout` `errorsThreshold`​ に達してからサーキットがオープンのままである期間。 - `circuitName` この設定にバインドするサーキットブレーカーの名前。デフォルトでは、各キューに独自のサーキットブレーカーがあります。 #### [](#グローバルサーキットブレーカー)グローバルサーキットブレーカー 各サブスクライバーが同じ「サーキット」の一部であるかのようにサーキットの状態を複数のサブスクライバーで共有する場合、グローバルサーキットブレーカーを使用します。 1. Anypoint Studio で、キャンバスの ​**\[Global Elements (グローバル要素)\]**​ タブをクリックします。 2. **\[Create (作成)\] > \[Component Configuration (コンポーネント設定)\] > \[Circuit Breaker (サーキットブレーカー)\]**​ を選択します。 ![「Global Element Properties (グローバル要素のプロパティ)」 ウィンドウのサーキットブレーカー設定のプロパティ](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-cb-global-canvas.png) Figure 6. このスクリーンショットは、​**\[Global Element Properties (グローバル要素のプロパティ)\]**​ ウィンドウのサーキットブレーカー設定のプロパティを示しています。 設定ウィザードで、必要に応じて次の項目を入力します。設定が完了したら、Anypoint MQ サブスクライバーからこの ​`Circuit_breaker`​ 宣言を参照できます。 サーキットブレーカーを参照する手順は、次のとおりです。 1. キャンバスで Subscriber ソースを選択します。 2. **\[Advanced (詳細)\]**​ タブをクリックします。 3. **\[Circuit Breaker (サーキットブレーカー)\] > \[Global Reference (グローバル参照)\]**​ を選択し、リストからグローバルサーキットブレーカー設定を選択します。 ![「Subscriber (サブスクライバー)」 プロパティウィンドウの 「Advanced (詳細)」 タブのサーキットブレーカーグローバル参照](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-cb-global-ref.png) Figure 7. 矢印は、\[Subscriber (サブスクライバー)\] プロパティウィンドウの ​**\[Advanced (詳細)\]**​ タブのサーキットブレーカーグローバル参照を示しています。 #### [](#非公開サーキットブレーカー)非公開サーキットブレーカー 非公開サーキットブレーカーは、1 つのサブスクライバーで内部的に宣言します。このサーキットの宣言は、Subscriber ソースが宣言されているフローでみ使用され、他のすべてのサーキットから分離されています。 この設定を使用する手順は、次のとおりです。 1. キャンバスで Subscriber ソースを選択します。 2. **\[Advanced (詳細)\]**​ タブをクリックします。 3. **\[Circuit breaker (サーキットブレーカー)\] > \[Edit Inline (インライン編集)\]**​ を選択し、項目に入力します。 ![「Subscriber (サブスクライバー)」 プロパティウィンドウの 「Advanced (詳細)」 タブで 「Edit inline (インライン編集)」 が選択されたサーキットブレーカー](/mulesoft/current/ja-jp/anypoint-mq-connector/3.x/_images/amq-3x-cb-inline.png) Figure 8. 矢印は、\[Subscriber (サブスクライバー)\] プロパティウィンドウの ​**\[Advanced (詳細)\]**​ タブで ​**\[Edit inline (インライン編集)\]**​ が選択されたサーキットブレーカーを示しています。 ### [](#サーキットブレーカーの例)サーキットブレーカーの例 #### [](#1-つのサブスクライバーのサーキット設定)1 つのサブスクライバーのサーキット設定 この例では、1 つのサブスクライバーがキューからメッセージをコンシュームし、REST API を使用して別のサービスにメッセージを投稿します。外部サービスに 5 回要求してタイムアウトした後、メッセージの処理を停止できます。処理が停止したら、Subscriber ソースは新しいメッセージで再試行する前に、サービスの回復を 30 秒待機します。 この例では、設定で次のサーキットブレーカーパラメーターを指定する必要があります。 ```xml errorsThreshold="5" tripTimeout="30" tripTimeoutUnit="SECONDS"/> ```

(1)

サーキットをトリップさせるエラー種別を設定します。エラーが ​errorsThreshold​ の回数発生すると、ポーリングが停止されます。

(2)

サーキットが障害状態にあるとみなされるために連続して発生する必要があるメッセージ数のしきい値を設定します。

(3)

errorsThreshold​ に達したためにサーキットブレーカーがトリップした後、新しいメッセージのポーリングを再開する前に待機する期間を指定します。

(4)

onErrorTypes​ パラメーターで予期されるエラーをスローする操作を定義します。

サーキットブレーカーでは、​`onErrorTypes`​ パラメーターにリストされていないすべてのエラーが無視されます。この例では、​`HTTP:BAD_REQUEST`​ などのエラーがサーキットブレーカーで無視されます。 #### [](#さまざまなキューからのサーキットの共有)さまざまなキューからのサーキットの共有 多くの場合、1 つの共通サービスでさまざまなキューからのメッセージを処理します。この例では、両方のサブスクライバーを 1 つのサーキットにバインドするよう ​`circuitName`​ パラメーターを設定します。 ```xml onErrorTypes="FTP:RETRY_EXHAUSTED, HTTP:SERVICE_UNAVAILABLE" errorsThreshold="10" tripTimeout="5" tripTimeoutUnit="MINUTES"/> config-ref="ftp-config"/> path="/external"/> ```

(1)

共通のサーキットブレーカーを複数のキューで共有するよう ​name​ パラメーターを設定します。

(2)

Subscriber ソースからのメッセージの処理に影響する可能性がある 2 つのエラーを識別し、それぞれ CSV リストとして渡します。

(3)

両方のサブスクライバーで、グローバルサーキットブレーカー設定を参照します。

(4)

複数のエラーがスローされる場合もありますが、サーキットブレーカーに関連しているのは ​FTP:RETRY_EXHAUSTED​ のみです。

(5)

HTTP Connector は ​HTTP:SERVICE_UNAVAILABLE​ エラーをスローすることがあり、その場合はメッセージの処理が阻止されます。

このシナリオの場合、両方のサブスクライバーで ​`FTP:RETRY_EXHAUSTED`​ および ​`HTTP:SERVICE_UNAVAILABLE`​ の両方のエラーを数え、その数が ​`errorsThreshold="10"`​ 値に達すると直ちにメッセージのポーリングが停止されます。​`tripTimeout`​ 値が経過すると、一方のサブスクライバーでメッセージをポーリングし、そのメッセージを使用してサーキットをテストします。そのメッセージの処理が成功すれば、両方のサブスクライバーでポーリングが有効になります。 ## [](#fifo-キュー)FIFO キュー FIFO キューは、単一コンシューマーのシナリオに最適です。1 つのコンシューマーがメッセージにアクセスしている場合、その他すべてのコンシューマーは最初のバッチが処理されるまでブロックされます。すべてのインフライトメッセージが肯定応答されるか否定応答されるまで、メッセージは配信されません。 メッセージグループを使用すると、複数のコンシューマーが FIFO キュー内のメッセージに同時にアクセスできます。この場合、あるコンシューマーがグループ内のメッセージにアクセスしているときに、別のコンシューマーは別のグループ内のメッセージにアクセスします。メッセージの順序は、各メッセージグループ内で保持されます。 FIFO キューでは、メッセージグループ ID でメッセージを取得することはできません。
メッセージの処理順序を保持するには、FIFO キューからのメッセージをコンシュームするフローで ​maxConcurrency​ の値を ​1​ に設定します。
### [](#fifo-キューとクラスタリング)FIFO キューとクラスタリング FIFO キューは、クラスター環境と非クラスター環境で同様に動作します。 FIFO キューは、指定された順序でメッセージをコンシュームします。メッセージがコンシュームされたら、その後のメッセージ処理を他のノードに分散できます。この場合、コンシューマーがメッセージを完全に処理する前に肯定応答すると、メッセージ処理中にメッセージの順序が変更される可能性があります。 - [オンプレミスの高可用性クラスタリング環境](https://docs.mulesoft.com/on-premises/latest/mule-high-availability-ha-clusters)​では、デフォルトで Anypoint MQ Connector の Subscriber ソースがすべてのノードで実行されます。 **\[Advanced (詳細)\]**​ タブの ​**\[Primary node only (プライマリノードのみ)\]**​ を選択して、プライマリノードとして実行するように動作を変更できます。 - 複数のワーカーを使用する ​[CloudHub](https://docs.mulesoft.com/cloudhub/)​ では、すべてのワーカーがプライマリノードとして実行されます。 この場合、アプリケーションを実行しているすべてのワーカーが同じ FIFO キューからコンシュームします。 ## [](#関連情報)関連情報 - [Anypoint MQ Consume 操作](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-consume) - [Anypoint MQ ACK および NACK 操作](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-ack) - [Anypoint MQ Connector リファレンス](https://docs.mulesoft.com/jp/anypoint-mq-connector/3.x/anypoint-mq-connector-reference) # Anypoint MQ Subscriber ソース Anypoint MQ Connector の Subscriber ソースは、アプリケーションで新しいメッセージをリスンし、宛先に到達したメッセージをコンシュームできます。さまざまなリスニング戦略を設定し、パフォーマンス、予測可能性、スケジュールに合わせてコンシュームを調整できます。 ## [](#最大メッセージスループット)最大メッセージスループット デフォルトでは、Subscriber ソースは継続的リスニングモードで機能し、キューに到達するとすぐにメッセージをコンシュームしてスループットを最大化します。これは、​`prefetch`​ subscriber-type 設定と呼ばれます。​`prefetch`​ モードを使用する場合、Subscriber ソースはメッセージのローカルバッファが常にフルになるように試みます。これにより、アプリケーションでメッセージを受け入れ可能になり次第、Mule フローにディスパッチできます。 ![\[Subscriber type (サブスクライバー種別)\] が \[Prefetch (プリフェッチ)\] に設定されている Subscriber 設定ウィンドウ](_images/amq-4x-subscriber-prefetch.png) Unresolved include directive in modules/ROOT/pages/anypoint-mq-listener.adoc - include::connectors::partial$mq-disclaimer.adoc\[\] `prefetch`​ subscriber-type 設定で、​`maxLocalMessages`​ パラメーターを設定してフルバッファの対象サイズを指定し、どれだけ迅速にキューからメッセージが取得されるのかを制御します。 ```xml ``` - より大きなバッファサイズを使用してパフォーマンスを最大化する。他のコンシューマーはローカルで保持されるメッセージを使用できないことに留意してください。 コンシューマーがキューからメッセージを取得してローカルバッファに保存すると、メッセージはインフライトメッセージとして他のコンシューマーに表示され、他のコンシューマーはコンシュームできなくなります。メッセージは、アプリケーションで処理を開始するまで必要に応じてこのインフライト状態のままになります。 - より小さなバッファサイズを使用して、キューにパブリッシュされたときにメッセージを処理する。 バッファを小さくすると、アプリケーションレベルのスループットが制限され、競合するコンシューマーがブロックされることを回避できます。 Unresolved include directive in modules/ROOT/pages/anypoint-mq-listener.adoc - include::connectors::partial$mq-disclaimer.adoc\[\] ### [](#fifo-and-prefetch)FIFO キューとプリフェッチモード `prefetch`​ モードを使用する場合、​`acknowledgementTimeout`​ タイマーは、メッセージがキューから取得されたときに開始されます。 Anypoint MQ は、​`acknowledgementTimeout`​ で指定された有効期限 (時間) を順守しようとします。 ただし、メッセージがバッファに保持されている時間がタイムアウト値の 80% に達すると、Anypoint MQ は有効期限タイムアウトを延長します。 Unresolved include directive in modules/ROOT/pages/anypoint-mq-listener.adoc - include::connectors::partial$mq-disclaimer.adoc\[\] ## [](#予測可能なメッセージのコンシューム)予測可能なメッセージのコンシューム アプリケーションで最大スループットではなく、メッセージのコンシュームを予測して制御する必要がある場合、ポーリングソースとして Subscriber を設定できます。Subscriber ソースは、固定スケジュールレートでキューの新規メッセージをチェックします。ポーリングがトリガーされるたびに、Subscriber ソースは 1 ~ 10 個のメッセージを取得して、フローに 1 つずつ順番にディスパッチします。 メッセージは、すぐにコンシュームされません。代わりに、Subscriber ソースの次のポーリングがトリガーされるまでキューに保持されます。キューには、取得されるメッセージが最大 10 個含まれます。たとえば、キューに 15 個のメッセージがある場合、1 つのポーリングで最初の 10 個のメッセージがコンシュームされ、残りの 5 個のメッセージは次のポーリングを待機します。 Subscriber をポーリングソースとして設定するには、​`polling`​ subscriber-type を指定します。 デフォルトの ​`polling`​ subscriber-type の宣言は次のようになります。 ![\[Subscriber type (サブスクライバー種別)\] が \[Polling (ポーリング)\] に設定されている Subscriber 設定ウィンドウ](_images/amq-4x-subscriber-polling.png) ```xml ``` デフォルトの ​`polling`​ subscriber-type の宣言では、宣言された宛先から 1 秒ごとにメッセージ 10 個の取得を試み、取得したメッセージを ​`MuleMessage`​ インスタンスとして 1 つずつフローにディスパッチします。 結果の ​`MuleMessage`​ には、以下が含まれます。 - メッセージのペイロードとしての本文 - メッセージ属性に含まれるメッセージのメタデータ ![ペイロードとメッセージ属性を含む Mule メッセージ出力](/mulesoft/current/ja-jp/anypoint-mq-connector/latest/_images/amq-4x-subscriber-output.png) 標準 Mule Runtime Engine スケジュール戦略を使用して、Subscriber ソースのポーリング戦略をカスタマイズできます。 - [一定間隔](https://docs.mulesoft.com/mule-runtime/latest/scheduler-concept#fixed-frequency) - [Cron 式](https://docs.mulesoft.com/mule-runtime/latest/scheduler-concept#cron-expressions) ### [](#カスタマイズされた固定ポーリング)カスタマイズされた固定ポーリング Subscriber ソースの新規メッセージのポーリング頻度はカスタマイズできます。これを行うには、カスタマイズされた ​`fixed-frequency`​ スケジュール戦略を使用して、​`polling`​ subscriber-type 戦略を宣言します。 デフォルトの ​`fixed-frequency`​ スケジュール戦略宣言は次のようになります。 ![\[Scheduling Strategy (スケジュール戦略)\] が \[Fixed Frequency (一定間隔)\] に設定されている Subscriber 設定ウィンドウ](_images/amq-4x-subscriber-polling-fixed.png) ```xml ``` ### [](#cron-ベースのポーリング)cron ベースのポーリング Anypoint MQ サブスクライバーは、​`cron`​ スケジュール戦略を使用することもできます。このスケジュール戦略を使用すると、「毎日午後 2 時から 2 時 59 分まで、毎分実行」のようなジョブのスケジュールが可能になります。このスケジュール戦略を使用するには、カスタマイズされた ​`cron`​ スケジュール戦略を使用して ​`polling`​ subscriber-type 戦略を宣言します。 ![\[Scheduling Strategy (スケジュール戦略)\] が \[Cron\] に設定されている Subscriber 設定ウィンドウ](_images/amq-4x-subscriber-polling-cron.png) ```xml ``` ## [](#メッセージ肯定応答)メッセージ肯定応答 フローの実行が正常に完了した場合や、処理前にアプリケーションがメッセージをコンシュームした場合にのみ受信したメッセージに自動的に肯定応答できます。 - **自動** デフォルトでは、Subscriber ソースは ​`AUTO`​ 肯定応答モードを使用します。このモードでは、メッセージフローの処理が成功した後で Subscriber ソースが取得するメッセージが自動的に肯定応答されます。つまり、Subscriber ソースはメッセージを取得してフローにディスパッチし、どのようにメッセージの処理が完了するのかを確認するために待機します。例外なしで処理が完了した場合にのみ ACK (肯定応答) を実行します。 例外が伝播されて処理フローの実行が完了すると、メッセージは自動的に肯定応答されず、再配信のためにキューに返されます。 詳細は、​[「自動肯定応答」](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack#automatic-acknowledgment)​を参照してください。 - **即時** `IMMEDIATE`​ 肯定応答モードを使用する場合、コンシュームされたメッセージが肯定応答 (キューから削除) され、処理のために Mule フローにディスパッチされます。 メッセージの肯定応答に失敗すると、メッセージは破棄されます。メッセージはフローにディスパッチされず、肯定応答がタイムアウトするまでインフライトのままになります。 Unresolved include directive in modules/ROOT/pages/anypoint-mq-listener.adoc - include::connectors::partial$mq-disclaimer.adoc\[\] `IMMEDIATE`​ 肯定応答モードでプリフェッチモードの Subscriber ソースを使用すると、スレッドが累積したり、アプリケーションが無応答になったりする場合があります。 詳細は、​[「即時肯定応答」](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack#immediate-acknowledgment)​を参照してください。 - **手動** `MANUAL`​ 肯定応答モードを使用する場合、​[ACK](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack)​ または ​[NACK](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack)​ ソースを使用して、メッセージの肯定応答を実行するタイミングをアプリケーションロジックで決定します。 手動肯定応答を実行するには、結果のメッセージ属性の一部として提供される ​`ackToken`​ の値が必要です。 詳細は、​[「手動肯定応答」](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack#manual-acknowledgment)​を参照してください。 肯定応答タイムアウトについての詳細は、​[「肯定応答タイムアウト」](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack#acknowledgment-timeout)​を参照してください。 ## [](#サーキットブレーカー機能)サーキットブレーカー機能 Subscriber ソースは、サーキットブレーク機能を提供します。この機能を使用すると、コンシュームされたメッセージの処理中に発生するエラーをコネクタでどのように処理するのかを制御できます。 たとえば、外部サービスに接続している場合、サーキットブレーカーを使用してそのサービスのダウンタイムを処理できます。サーキットブレーカーでは、要求の実行を停止したり、負荷を減らして外部サービスを回復したりできます。 Mule 4 アプリケーションでの Anypoint MQ の使用は、キューからメッセージをコンシュームし、外部サービスを使用して処理を行う、MQ サブスクライバーを使用する Mule フローがあることを意味します。このサービスが使用できない場合、次のようになります。 1. 要求が失敗する。 2. エラーが発生する。 3. メッセージの処理が失敗に終わるか、デッドレターキュー (DLQ) にメッセージを送信するなど、メッセージ処理のカスタムエラーで終了します。 外部サービスを利用できない場合、メッセージを処理するすべての試みは失敗し、アプリケーションは成功しないメッセージのコンシュームを繰り返し実行し続けます。この動作は、一定期間メッセージをさらにコンシュームしないように、サブスクライバーにエラーを通知すれば回避できます。 ### [](#サーキットブレーカープロセス)サーキットブレーカープロセス Subscriber ソースによって提供されるサーキットブレーカー機能は、Mule で提供されるエラー処理メカニズムにバインドされています。エラー通知メカニズムを使用してサーキット障害と呼ばれる外部サービスに関連するエラー数を記録します。サーキット障害にはあらゆるエラーをバインドできます。たとえば、​`HTTP:TIMEOUT`​ や ​`FTP:SERVICE_NOT_AVAILABLE`​、さらには ​`ORG:EXTERNAL_ERROR`​ のようなアプリケーションのカスタムエラーもバインドできます。 Mule フローの実行がエラーに終わった場合、Subscriber ソースはエラーが外部サービスエラーを示す ​`onErrorTypes`​ のいずれかであるかどうかをチェックし、​`errorsThreshold`​ に達するまで、連続発生数を数えます。 `errorsThreshold`​ に達すると、サーキットがトリップし、​`tripTimeout`​ によって指定される期間中、オープン状態のままになります。 サーキットがオープン状態の間、ポーリングまたはコンシュームされたすべてのメッセージは肯定応答 (NACK) されず、Anypoint MQ は ​`tripTimeout`​ が経過するまでその後の操作 (ポーリング、ACK、NACK、パブリッシュ) を停止します。 `tripTimeout`​ が経過すると、サーキットブレーカーは新しい​[状態](#circuit-breaker-states)​に移行します。 新しい状態が ​**\[Closed (クローズ)\]**​ の場合、クライアントは次のポーリングでメッセージをコンシュームできます。 アプリケーションを CloudHub にデプロイするワーカーが 1 つか複数かによって、この動作は異なります。 1 つのワーカー ワーカーはメッセージをキューから個別に処理し、エラー数が ​`errorsThreshold`​ の値に達するとサーキットブレーカーをオープン状態にします。 複数のワーカー 各ワーカーに関連付けられたサーキットブレーカーがあり、​`errorsThreshold`​ と ​`tripTimeout`​ の値を個別に追跡します。 各ワーカーはメッセージをキューから個別に処理し、エラー数がそのワーカーで設定された ​`errorsThreshold`​ の値に達するとサーキットブレーカーをオープン状態にします。 デフォルトでは、サーキットブレーク機能は無効になっています。 ### [](#circuit-breaker-states)サーキットブレーカーの状態 サーキットブレーカーは、3 つの状態 (クローズド、オープン、ハーフオープン) があります。アプリケーションの動作は、現在の状態に基づいて変わります。詳細は、 [「Microsoft Circuit Breaker pattern (Microsoft サーキットブレーカーパターン)」](https://docs.microsoft.com/en-us/azure/architecture/patterns/circuit-breaker)​を参照してください。 ![サーキットブレーカーのクローズ、ハーフオープン、オープンの状態が表示されているフロー](/mulesoft/current/ja-jp/anypoint-mq-connector/latest/_images/amq-4x-subscriber-states-view.png) - **クローズド** 通常 Subscriber ソースが設定に基づいて MQ からメッセージを取得する開始状態。実質的にサーキットブレーカーが存在しないかのように機能します。 - **クローズドからオープンへの遷移** メッセージの処理中に成功することなく失敗が連続して発生し、​`errorsThreshold`​ 値に達すると、サーキットブレーカーがトリップし、オープン状態に遷移します。 フローにすでにディスパッチされたメッセージは、結果が成功か失敗かに関係なく処理を完了します。 ブローカーのインフライトであるが、まだディスパッチされておらず、ローカルに保持されているメッセージは、肯定応答されず、別のコンシューマーに再配信するためにキューに戻されます。 - **オープン** Subscriber ソースはメッセージの取得を試みず、​`tripTimeout`​ に達するまで通知せずにイテレーションをスキップします。 - **ハーフオープン** `tripTimeout`​ の経過後、Subscriber ソースはハーフオープン状態になります。次回のメッセージのポーリングで、Subscriber ソースは通常のクローズド状態に戻る前にサービスからメッセージを 1 つ取得し、そのメッセージを使用してシステムが回復したかどうかをチェックします。 Subscriber ソースが 1 つのメッセージを正常にフェッチしてフローにディスパッチし、処理が正常に完了すると、Subscriber ソースは通常の状態に戻り、すぐに追加のメッセージのフェッチを試みます。 Mule フローの処理が、予期される ​`onErrorTypes`​ のいずれかで失敗すると、サーキットはオープン状態に戻り、​`tripTimeout`​ タイマーがリセットされます。 ### [](#サーキットブレーカーの設定)サーキットブレーカーの設定 サーキットブレーカーは、グローバルサーキットブレーカーまたは非公開サーキットブレーカーとして設定できます。 どちらの方法でも、設定パラメーターは同じです。 - `onErrorTypes` フローの実行中に障害とみなされるエラー種別。エラーの発生は、フローがエラーの伝達によって完了した場合にのみ計数されます。デフォルトでは、すべてのエラーがサーキット障害とみなされます。 - `errorsThreshold` サーキットブレーカーがオープンになるのに発生する必要がある ​`onErrorTypes`​ エラーの回数。 - `tripTimeout` `errorsThreshold`​ に達してからサーキットがオープンのままである期間。 - `circuitName` この設定にバインドするサーキットブレーカーの名前。デフォルトでは、各キューに独自のサーキットブレーカーがあります。 #### [](#グローバルサーキットブレーカー)グローバルサーキットブレーカー 各サブスクライバーが同じ「サーキット」の一部であるかのようにサーキットの状態を複数のサブスクライバーで共有する場合、グローバルサーキットブレーカーを使用します。 1. Anypoint Studio で、キャンバスの ​**\[Global Elements (グローバル要素)\]**​ タブをクリックします。 2. **\[Create (作成)\] > \[Component Configuration (コンポーネント設定)\] > \[Circuit Breaker (サーキットブレーカー)\]**​ を選択します。 ![エラー種別、エラーしきい値、トリップタイムアウト、トリップタイムアウト単位を含むサーキットブレーカー設定ウィンドウ](/mulesoft/current/ja-jp/anypoint-mq-connector/latest/_images/amq-4x-cb-global-canvas.png) 設定ウィザードで、必要に応じて次の項目を入力します。設定が完了したら、Anypoint MQ サブスクライバーからこの ​`Circuit_breaker`​ 宣言を参照できます。 サーキットブレーカーを参照する手順は、次のとおりです。 1. キャンバスで Subscriber ソースを選択します。 2. **\[Advanced (詳細)\]**​ タブをクリックします。 3. **\[Circuit Breaker (サーキットブレーカー)\] > \[Global Reference (グローバル参照)\]**​ を選択し、リストからグローバルサーキットブレーカー設定を選択します。 ![\[Global reference (グローバル参照)\] が \[Circuit\_breaker\] に設定されている Subscriber 設定ウィンドウ](_images/amq-4x-cb-global-ref.png) #### [](#非公開サーキットブレーカー)非公開サーキットブレーカー 非公開サーキットブレーカーは、1 つのサブスクライバーで内部的に宣言します。このサーキットの宣言は、Subscriber ソースが宣言されているフローでみ使用され、他のすべてのサーキットから分離されています。 この設定を使用する手順は、次のとおりです。 1. キャンバスで Subscriber ソースを選択します。 2. **\[Advanced (詳細)\]**​ タブをクリックします。 3. **\[Circuit breaker (サーキットブレーカー)\] > \[Edit Inline (インライン編集)\]**​ を選択し、項目に入力します。 ![\[Circuit breaker (サーキットブレーカー)\] が \[Edit Inline (インライン編集)\] に設定されている Subscriber 設定ウィンドウ](_images/amq-4x-cb-inline.png) ### [](#サーキットブレーカーの例)サーキットブレーカーの例 #### [](#1-つのサブスクライバーのサーキット設定)1 つのサブスクライバーのサーキット設定 この例では、1 つのサブスクライバーがキューからメッセージをコンシュームし、REST API を使用して別のサービスにメッセージを投稿します。外部サービスに 5 回要求してタイムアウトした後、メッセージの処理を停止できます。処理が停止したら、Subscriber ソースは新しいメッセージで再試行する前に、サービスの回復を 30 秒待機します。 この例では、設定で次のサーキットブレーカーパラメーターを指定する必要があります。 ```xml errorsThreshold="5" tripTimeout="30" tripTimeoutUnit="SECONDS"/> ```

(1)

サーキットをトリップさせるエラー種別を設定します。エラーが ​errorsThreshold​ の回数発生すると、ポーリングが停止されます。

(2)

サーキットが障害状態にあるとみなされるために連続して発生する必要があるメッセージ数のしきい値を設定します。

(3)

errorsThreshold​ に達したためにサーキットブレーカーがトリップした後、新しいメッセージのポーリングを再開する前に待機する期間を指定します。

(4)

onErrorTypes​ パラメーターで予期されるエラーをスローする操作を定義します。

サーキットブレーカーでは、​`onErrorTypes`​ パラメーターにリストされていないすべてのエラーが無視されます。この例では、​`HTTP:BAD_REQUEST`​ などのエラーがサーキットブレーカーで無視されます。 #### [](#さまざまなキューからのサーキットの共有)さまざまなキューからのサーキットの共有 多くの場合、1 つの共通サービスでさまざまなキューからのメッセージを処理します。この例では、両方のサブスクライバーを 1 つのサーキットにバインドするよう ​`circuitName`​ パラメーターを設定します。 ```xml onErrorTypes="FTP:RETRY_EXHAUSTED, HTTP:SERVICE_UNAVAILABLE" errorsThreshold="10" tripTimeout="5" tripTimeoutUnit="MINUTES"/> config-ref="ftp-config"/> path="/external"/> ```

(1)

共通のサーキットブレーカーを複数のキューで共有するよう ​name​ パラメーターを設定します。

(2)

Subscriber ソースからのメッセージの処理に影響する可能性がある 2 つのエラーを識別し、それぞれ CSV リストとして渡します。

(3)

両方のサブスクライバーで、グローバルサーキットブレーカー設定を参照します。

(4)

複数のエラーがスローされる場合もありますが、サーキットブレーカーに関連しているのは ​FTP:RETRY_EXHAUSTED​ のみです。

(5)

HTTP Connector は ​HTTP:SERVICE_UNAVAILABLE​ エラーをスローすることがあり、その場合はメッセージの処理が阻止されます。

このシナリオの場合、両方のサブスクライバーで ​`FTP:RETRY_EXHAUSTED`​ および ​`HTTP:SERVICE_UNAVAILABLE`​ の両方のエラーを数え、その数が ​`errorsThreshold="10"`​ 値に達すると直ちにメッセージのポーリングが停止されます。​`tripTimeout`​ 値が経過すると、一方のサブスクライバーでメッセージをポーリングし、そのメッセージを使用してサーキットをテストします。そのメッセージの処理が成功すれば、両方のサブスクライバーでポーリングが有効になります。 ## [](#fifo-キュー)FIFO キュー FIFO キューは、単一コンシューマーのシナリオに最適です。1 つのコンシューマーがメッセージにアクセスしている場合、その他すべてのコンシューマーは最初のバッチが処理されるまでブロックされます。すべてのインフライトメッセージが肯定応答されるか否定応答されるまで、メッセージは配信されません。 メッセージグループを使用すると、複数のコンシューマーが FIFO キュー内のメッセージに同時にアクセスできます。この場合、あるコンシューマーがグループ内のメッセージにアクセスしているときに、別のコンシューマーは別のグループ内のメッセージにアクセスします。メッセージの順序は、各メッセージグループ内で保持されます。 FIFO キューでは、メッセージグループ ID でメッセージを取得することはできません。
メッセージの処理順序を保持するには、FIFO キューからのメッセージをコンシュームするフローで ​maxConcurrency​ の値を ​1​ に設定します。
### [](#fifo-キューとクラスタリング)FIFO キューとクラスタリング FIFO キューは、クラスター環境と非クラスター環境で同様に動作します。 FIFO キューは、指定された順序でメッセージをコンシュームします。メッセージがコンシュームされたら、その後のメッセージ処理を他のノードに分散できます。この場合、コンシューマーがメッセージを完全に処理する前に肯定応答すると、メッセージ処理中にメッセージの順序が変更される可能性があります。 - [オンプレミスの高可用性クラスタリング環境](https://docs.mulesoft.com/on-premises/latest/mule-high-availability-ha-clusters)​では、デフォルトで Anypoint MQ Connector の Subscriber ソースがすべてのノードで実行されます。 **\[Advanced (詳細)\]**​ タブの ​**\[Primary node only (プライマリノードのみ)\]**​ を選択して、プライマリノードとして実行するように動作を変更できます。 - 複数のワーカーを使用する ​[CloudHub](https://docs.mulesoft.com/cloudhub/)​ では、すべてのワーカーがプライマリノードとして実行されます。 この場合、アプリケーションを実行しているすべてのワーカーが同じ FIFO キューからコンシュームします。 ## [](#関連情報)関連情報 - [Anypoint MQ Consume 操作](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-consume) - [Anypoint MQ ACK および NACK 操作](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-ack) - [Anypoint MQ Connector リファレンス](https://docs.mulesoft.com/jp/anypoint-mq-connector/latest/anypoint-mq-connector-reference) # アプリケーションでのクロスリージョンフェールオーバーの設定 Mule 4 用 Anypoint MQ Connector バージョン 4.0.7 以降では、標準キューのフェールオーバー機能が提供されます。 この機能はデフォルトでは有効化されていません。 この機能を使用する前に、[mq::mq-failover.adoc](#mq::mq-failover.adoc)を参照してください。
キューのフェールオーバーを有効にしてプライマリリージョンがダウンすると、プライマリリージョンが使用可能になるまで、データは国外のリージョンに存在する可能性があります。
フェールオーバー機能を使用する手順は、次のとおりです。 - アプリケーションの場合 1. Anypoint MQ Connector をバージョン 4.0.7 以降にアップグレードします。 2. [アプリケーションでフェールオーバーを有効にします](#enable-failover-feature)​。 3. (省略可能) ​[チェック頻度を変更します](#configure-check-frequency)​。 - Anypoint MQ の場合 1. キューを作成または変更し、​[キューのフェールオーバーを有効にします](#mq::mq-failover.adoc#enable-failover)​。 ## [](#enable-failover-feature)アプリケーションでフェールオーバーを有効にする アプリケーションでフェールオーバー機能を有効にするには、​`cross.region.failover.feature.enabled`​ システムプロパティを ​`true`​ に設定します。 このプロパティは、デフォルトでは ​`false`​ に設定されています。
このフェールオーバープロパティが有効になっている場合、コネクタはプライマリキューとフェイーオーバーキューの両方の処理メッセージを処理します。 中断を防ぐため、アプリケーションは代替キューから直接パプリッシュやコンシュームを行うことはできません。 詳細は、​代替キューから直接メッセージを処理する​を参照してください。
Runtime Manager でこのアプリケーションプロパティを設定する手順は、次のとおりです。 1. Anypoint Platform にサインインし、ナビゲーションメニューで ​**\[Runtime Manager\]**​ をクリックします。 2. 新しいアプリケーションをデプロイするか、デプロイ済みアプリケーションの設定を変更します。 3. **\[Properties (プロパティ)\]**​ タブで、プロパティの値を指定します。次に例を示します。 ![cross.region.failover.feature.enabled が true に設定された \[Properties (プロパティ)\] テーブル](_images/mq-set-properties-failover-enabled.png) または、Anypoint Studio の ​**\[Run Configurations (設定を実行)\]**​ でプロパティを設定します。 詳細は、​[「Mule アプリケーションでのシステムプロパティ」](https://docs.mulesoft.com/mule-runtime/latest/mule-app-properties-system)​を参照してください。 ## [](#consume-directly-fallback)代替キューから直接メッセージを処理する アプリケーションでのフェールオーバーが有効になっている場合、Anypoint MQ Connector は代替キュー (​`_fb`​ サフィックス付き) を管理し、代替キューで直接パブリッシュ、コンシューム、またはサブスクライブ操作を実行できないようにします。 アプリケーションが代替キューに直接接続すると、Anypoint MQ Connector はエラーを返します。 ```none java.lang.IllegalArgumentException: Using fallback queue directly is not supported ``` (Anypoint MQ UI から) キューのフェールオーバーを無効にしてアプリケーションを再起動すると、コネクタは代替キューからパブリッシュまたはコンシュームしなくなります。 フェールオーバーを無効にした後に代替キューに残っているメッセージを処理する手順は、次のとおりです。 1. 代替キューから直接処理するようにアプリケーションを設定します。 Consume または Subscriber 操作の場合、​**\[Queue (キュー)\]**​ 項目を代替キューの名前に設定します。次に例を示します。 ![キュー名 myDemoQueue\_fb を含む Subscriber 設定](/mulesoft/current/ja-jp/anypoint-mq-connector/latest/_images/mq-publish-fallback-queue-subscribe.png) 2. アプリケーションが代替キューから直接メッセージをコンシュームできるようにします。 1. Anypoint Platform にサインインし、ナビゲーションメニューで ​**\[Runtime Manager\]**​ をクリックします。 2. デプロイ済みアプリケーションの設定を変更します。 3. **\[Properties (プロパティ)\]**​ タブで、​`cross.region.failover.feature.enabled`​ システムプロパティを ​`false`​ に設定します。次に例を示します。 ![cross.region.failover.feature.enabled が false に設定された \[Properties (プロパティ)\] テーブル](_images/mq-set-properties-failover-disabled.png) または、Anypoint Studio の ​**\[Run Configurations (設定を実行)\]**​ でプロパティを設定します。 詳細は、​[「Mule アプリケーションでのシステムプロパティ」](https://docs.mulesoft.com/mule-runtime/latest/mule-app-properties-system)​を参照してください。 ## [](#configure-check-frequency)代替キューチェックの頻度を設定する デフォルトでは、プライマリリージョンがダウンしている場合、Anypoint MQ Connector は次のチェックを行います。 - 新規メッセージの場合、5 分 (300,000 ミリ秒) ごとに 1 回代替キューをチェック - 1 分 (60,000 ミリ秒) ごとに 1 回プライマリリージョンがバックアップされているかどうかをプライマリキューでチェック この動作を変更するには、次のプロパティを設定します。
プロパティ 説明 最小値 最大値 デフォルト値

primary.region.status.check.interval.ms

代替リージョンからの処理時にコネクタがプライマリリージョンの可用性をチェックする頻度

1 分 (60,000 ミリ秒)

15 分 (900,000 ミリ秒)

1 分 (60,000 ミリ秒)

fallback.messages.check.interval.ms

コネクタが受信メッセージの代替キューをチェックする頻度

1 分 (60,000 ミリ秒)

12 時間 (4,320 万ミリ秒)

5 分 (300,000 ミリ秒)

Runtime Manager でこれらのアプリケーションプロパティを設定する手順は、次のとおりです。 1. Anypoint Platform にサインインし、ナビゲーションメニューで ​**\[Runtime Manager\]**​ をクリックします。 2. 新しいアプリケーションをデプロイするか、デプロイ済みアプリケーションの設定を変更します。 3. **\[Properties (プロパティ)\]**​ タブで、次のプロパティの値 (ミリ秒) を指定します。 ![プライマリおよび代替状況チェック間隔を含む \[Properties (プロパティ)\] テーブル](_images/mq-set-properties-failover.png) または、Anypoint Studio の ​**\[Run Configurations (設定を実行)\]**​ でこれらのプロパティを設定します。 詳細は、​[「Mule アプリケーションでのシステムプロパティ」](https://docs.mulesoft.com/mule-runtime/latest/mule-app-properties-system)​を参照してください ### [](#check-interval)チェック間隔の考慮事項 チェック間隔を微調整するためにアプリケーションプロパティ値を設定する方法を決定する場合、次の点を考慮してください。 - プライマリリージョンのダウンタイム中に代替リージョンにパブリッシュする場合、Anypoint MQ Connector の Publish 操作は、デフォルトでプライマリリージョンが 1 分に 1 回バックアップされているかどうかを確認します。 フローが Publish 操作をトリガーする頻度が 1 分に 1 回未満の場合、コネクタはこの操作がトリガーされるたびにプライマリリージョンがバックアップされているかどうかを確認します。 この動作を微調整するには、​`primary.region.status.check.interval.ms`​ の値を、フローが Publish 操作をトリガーする頻度より大きい値に設定します。 - フローが Anypoint MQ Connector の Consumer または Subscriber 操作をトリガーすると、コネクタはプライマリキューをチェックする前に、代替キューで新規メッセージをチェックします (デフォルトでは 5 分ごと)。 フローが Consumer または Subscriber 操作をトリガーする頻度が 5 分ごとよりも短い場合、コネクタはプライマリキューからコンシュームする前に、代替キューで新規メッセージを確認します。 代替キューをチェックする頻度を減らすには、​`fallback.messages.check.interval.ms`​ の値を、フローが Consumer または Subscriber 操作をトリガーする頻度より大きい値に設定します。
このプロパティを高い値に設定すると、代替キュー内のメッセージは頻繁に処理されないため、メッセージは順不同になる可能性があります。
## [](#関連情報)関連情報 - [mq::mq-failover.adoc](#mq::mq-failover.adoc) - [CloudHub 2.0 でのプロパティを使用したアプリケーションの動作の変更](https://docs.mulesoft.com/cloudhub-2/ch2-manage-props) - [Manage Properties for Applications on CloudHub](https://docs.mulesoft.com/cloudhub/cloudhub-manage-props) - [Anypoint Exchange の Anypoint MQ Connector](https://anypoint.mulesoft.com/exchange/com.mulesoft.connectors/anypoint-mq-connector/) - [Anypoint MQ Connector Release Notes for Mule 4](https://docs.mulesoft.com/jp/release-notes/connector/anypoint-mq-connector-release-notes-mule-4)