---
schema_version: "1"
id: 3efdd60fbc434982f8746b48d9f989d23d23c02334902b3a5fca1a42f56403ed
title: VM Connector を使用したキュー内保留メッセージの集約の例
collection: mulesoft
version: current
locale: ja-jp
portal_url: https://docs.salesforce.com/mulesoft/current/ja-jp/vm-connector/latest/vm-aggregate-messages.md
canonical_url: https://docs.mulesoft.com/jp/vm-connector/latest/vm-aggregate-messages
source_site: docs.mulesoft.com
content_type: documentation
content_hash: dbf10bc512c8fb7215fef206f250f71647a9e1b405d09505ad2b8b059dff2808
description: MuleSoft Documentation Site
release: "2.0"
latest: true
---
# VM Connector を使用したキュー内保留メッセージの集約の例

次の例は、仮想マシン用 Anypoint Connector (VM Connector) を使用して、キューで保留中のメッセージを集約する方法についての 2 つのシナリオを示しています。

VM はキューであり、キュー内で保留中のすべてのメッセージに対して for each 操作を実行することはできません。

## [](#キュー内メッセージ集計の最初の例)キュー内メッセージ集計の最初の例

この例では、Aggregators Module を使用して、VM ​**Listener**​ 操作によって受信されたメッセージを集約し、集約の結果を反復処理します。

![Anypoint Studio キャンバスのキュー内保留メッセージの集約の最初のフロー](/mulesoft/current/ja-jp/vm-connector/latest/_images/vm-aggregate-flow-1.png)

### [](#最初のフローの作成)最初のフローの作成

最初のフローでは、HTTP ​**\[Listener\]**​ ソースが受信 HTTP メッセージをリスンします。次に、VM ​**Publish**​ 操作が VM キュー内の特定のコンテンツをパブリッシュします。

1. Studio で、新しい Mule プロジェクトを作成します。
2. **\[Mule Palette (Mule パレット)\]**​ で、​**\[HTTP\] > \[Listener\]**​ ソースを選択してキャンバスにドラッグします。
3. HTTP ​**\[Listener\]**​ 設定画面で、​**\[Path (パス)\]**​ 項目を ​`/`​ に設定します。
4. **\[Connector configuration (コネクタ設定)\]**​ 項目の横にあるプラス記号 (+) をクリックして、アプリケーション内の HTTP ​**Listener**​ のすべてのインスタンスで使用できるグローバル要素を設定します。
5. HTTP グローバル設定の必須項目を入力します。
6. **\[OK\]**​ をクリックします。
7. **\[For Each\]**​ コンポーネントを HTTP の ​**\[Listener\]**​ ソースの右にドラッグします。
8. **\[Collection (コレクション)\]**​ を ​`1 to 200`​ に設定します。
9. **\[VM\] > \[Publish\]**​ 操作を ​**\[For Each\]**​ コンポーネントにドラッグします。
10. **\[Content (コンテンツ)\]**​ を ​`"hello world"`​ に設定します。
11. **\[Connector configuration (コネクタ設定)\]**​ 項目の横にあるプラス記号 (+) をクリックして、アプリケーション内の VM ​**Publish**​ 操作のすべてのインスタンスで使用できるグローバル要素を設定します。
12. **\[VM Config (VM 設定)\]**​ ウィンドウで、プラス記号 (+) をクリックして新しいキューを追加します。
13. **\[Queue name (キュー名)\]**​ を ​`demo`​ に設定します。
14. **\[Finish (完了)\]**​ をクリックします。
15. **\[OK\]**​ をクリックします。

### [](#2-つ目のフローの作成)2 つ目のフローの作成

2 番目のフローでは VM ​**\[Listener\]**​ 操作が VM キューにパブリッシュされたメッセージをリスンします。​**\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープを使用すると、時間条件を満たすまで受信したメッセージを集約することができます。

1. Studio で ​**\[VM\] > \[Listener\]**​ 操作を最初のフローの下にドラッグします。
2. **\[Connector configuration (コネクタ設定)\]**​ で、前に作成した ​`VM_Config`​ 設定を選択します。
3. **\[Queue name (キュー名)\]**​ を ​`demo`​ に設定します。
4. **\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープを ​**\[VM Listener\]**​ の右にドラッグします。
5. **\[Name (名前)\]**​ を ​`demoaggregator`​ に設定します。
6. **\[Period (期間)\]**​ を ​`5`​ に設定して、5 秒間に達するまで受信メッセージを集約します。

### [](#3-つ目のフローを作成する)3 つ目のフローを作成する

このフローでは、​**\[Aggregator listener\]**​ 操作は ​**\[Time based aggregator (時間ベースのアグリゲーター)\]**​ スコープによって集約されたメッセージをリスンします。​**\[For Each\]**​ コンポーネントはメッセージを反復処理し、​**\[Logger\]**​ コンポーネントは各メッセージのコンテンツを記録します。

1. Studio で、​**\[Aggregator listener\]**​ ソースを 2 番目のフローの上にドラッグします。
2. **\[Aggregator name (アグリゲーター名)\]**​ を ​`demoaggregator`​ に設定します。
3. **\[For Each\]**​ コンポーネントを ​**\[Aggregator listener\]**​ ソースの右にドラッグします。
4. **\[Collection (コレクション)\]**​ を ​`[#payload]`​ に設定します。
5. **\[Logger\]**​ コンポーネントを ​**\[For Each\]**​ コンポーネントにドラッグします。
6. **\[Message (メッセージ)\]**​ を ​`[#payload]`​ に設定します。
7. Mule アプリケーションを保存して実行します。
8. ターミナルで、curl コマンド ​`curl [http://localhost:8081/](http://localhost:8081/)`​ を実行します。

### [](#キュー内メッセージ集約の最初の例の-xml)キュー内メッセージ集約の最初の例の XML

この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。

```xml
<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:aggregators="http://www.mulesoft.org/schema/mule/aggregators" xmlns:http="http://www.mulesoft.org/schema/mule/http"
	xmlns:vm="http://www.mulesoft.org/schema/mule/vm"
	xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/vm http://www.mulesoft.org/schema/mule/vm/current/mule-vm.xsd
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
http://www.mulesoft.org/schema/mule/aggregators http://www.mulesoft.org/schema/mule/aggregators/current/mule-aggregators.xsd">
	<http:listener-config name="HTTP_Listener_config" basePath="/" >
		<http:listener-connection host="0.0.0.0" port="8081" />
	</http:listener-config>
	<vm:config name="VM_Config">
		<vm:connection />
		<vm:queues >
			<vm:queue queueName="demo" />
		</vm:queues>
	</vm:config>
	<flow name="demo-vm2Flow" >
		<http:listener config-ref="HTTP_Listener_config" path="/"/>
		<foreach collection="1 to 200">
			<vm:publish config-ref="VM_Config" queueName="demo" >
				<vm:content ><![CDATA[#["hello world"]]]></vm:content>
			</vm:publish>
		</foreach>
	</flow>
	<flow name="demo-vm2Flow2" >
		<aggregators:aggregator-listener aggregatorName="demoaggregator" includeTimedOutGroups="true"/>
		<foreach collection="#[payload]">
			<logger level="INFO" message="#[payload]"/>
		</foreach>
	</flow>
	<flow name="demo-vm2Flow1" >
		<vm:listener queueName="demo" config-ref="VM_Config"/>
		<aggregators:time-based-aggregator  name="demoaggregator" period="5">
		</aggregators:time-based-aggregator>
	</flow>
</mule>
```

## [](#キュー内メッセージ集計の-2-番目の例)キュー内メッセージ集計の 2 番目の例

この例では、​**\[For Each\]**​ コンポーネントは VM ​**\[Consume\]**​ 操作によってコンシュームされたメッセージを反復処理します。Mule アプリケーションは受信したすべてのメッセージを変数に追加し、各メッセージのペイロードコンテンツを記録します。この例のシナリオではキューにメッセージがなくなるまでメッセージを取得します。キューにメッセージがなくなると、VM ​**\[Consume\]**​ 操作はタイムアウトメッセージをスローし、メッセージは ​**\[On Error Continue\]**​ コンポーネントで処理され、結果が使用可能になります。

![Anypoint Studio キャンバスのキュー内保留メッセージの集約の 2 番目のフロー](/mulesoft/current/ja-jp/vm-connector/latest/_images/vm-aggregate-flow-2.png)

### [](#最初のフローの作成-2)最初のフローの作成

最初のフローでは ​**\[HTTP Listener\]**​ ソースが受信 HTTP メッセージをリスンし、VM ​**\[Publish\]**​ 操作が VM キューに特定のコンテンツをパブリッシュします。

1. Studio で、新しい Mule プロジェクトを作成します。
2. **\[Mule Palette (Mule パレット)\]**​ で、​**\[HTTP\] > \[Listener\]**​ ソースを選択してキャンバスにドラッグします。
3. HTTP ​**\[Listener\]**​ 設定画面で、​**\[Path (パス)\]**​ 項目を ​`/`​ に設定します。
4. **\[Connector configuration (コネクタ設定)\]**​ 項目の横にあるプラス記号 (+) をクリックして、アプリケーション内の HTTP ​**Listener**​ のすべてのインスタンスで使用できるグローバル要素を設定します。
5. HTTP グローバル設定の必須項目を入力します。
6. **\[OK\]**​ をクリックします。
7. **\[For Each\]**​ コンポーネントを HTTP の ​**\[Listener\]**​ ソースの右にドラッグします。
8. **\[Collection (コレクション)\]**​ を ​`1 to 55`​ に設定します。
9. **\[VM\] > \[Publish\]**​ 操作を ​**\[For Each\]**​ コンポーネントにドラッグします。
10. **\[Content (コンテンツ)\]**​ を ​`"test world"`​ に設定します。
11. **\[Connector configuration (コネクタ設定)\]**​ 項目の横にあるプラス記号 (+) をクリックして、アプリケーション内の VM ​**Publish**​ 操作のすべてのインスタンスで使用できるグローバル要素を設定します。
12. **\[VM Config (VM 設定)\]**​ ウィンドウで、プラス記号 (+) をクリックして新しいキューを追加します。
13. **\[Queue name (キュー名)\]**​ を ​`demo`​ に設定します。
14. **\[Finish (完了)\]**​ をクリックします。
15. **\[OK\]**​ をクリックします。

### [](#2-つ目のフローの作成-2)2 つ目のフローの作成

2 番目のフローでは VM ​**\[Consume\]**​ 操作がキューからパブリッシュされたメッセージを取得します。​**\[Transform Message\]**​ コンポーネントは、ペイロードコンテンツ変換し、変数に保存します。Mule アプリケーションは各変数ペイロードを記録します。

1. Studio で ​**\[Scheduler\]**​ ソースを最初のフローの下にドラッグします。
2. **\[Frequency (頻度)\]**​ を ​`5000`​ に設定します。
3. **\[Set Variable\]**​ コンポーネントを ​**\[Scheduler\]**​ コンポーネントの右にドラッグします。
4. **\[Name (名前)\]**​ を ​`result`​ に設定します。
5. **\[Value (値)\]**​ を ​`[]`​ に設定します。
6. **\[Try\]**​ スコープコンポーネントを ​**\[Set Variable\]**​ の右にドラッグします。
7. **\[For Each\]**​ コンポーネントを ​**\[Try\]**​ スコープにドラッグします。
8. **\[Collection (コレクション)\]**​ を ​`1 to 10000`​ に設定します。
9. VM ​**\[Consume\]**​ 操作を ​**\[For Each\]**​ コンポーネントにドラッグします。
10. **\[Connector configuration (コネクタ設定)\]**​ で、前に作成した ​`VM_Config`​ 設定を選択します。
11. **\[Queue name (キュー名)\]**​ を ​`demo`​ に設定します。
12. **\[Transform Message\]**​ コンポーネントを ​**\[Consume\]**​ の右にドラッグします。
13. DataWeave スクリプトを次のように設定します。

%dw 2.0
output application/java
---
vars.result << payload

1. **\[On Error Continue\]**​ コンポーネントを ​**\[Try\]**​ スコープコンポーネントの ​**\[Error handling (エラー処理)\]**​ セクションにドラッグします。
2. **\[Type (種別)\]**​ を ​`VM:EMPTY_QUEUE`​ に設定します。
3. **\[Set Payload\]**​ コンポーネントを ​**\[Try\]**​ スコープコンポーネントの右にドラッグします。
4. **\[Value (値)\]**​ を ​`vars.result`​ に設定します。
5. **\[For Each\]**​ コンポーネントを ​**\[Set Payload\]**​ コンポーネントの右にドラッグします。
6. **\[Collection (コレクション)\]**​ を ​`payload`​ に設定します。
7. **\[Logger\]**​ コンポーネントを ​**\[For Each\]**​ スコープコンポーネントにドラッグします。
8. **\[Message (メッセージ)\]**​ を ​`payload`​ に設定します。
9. Mule アプリケーションを保存して実行します。
10. ターミナルで、curl コマンド ​`curl [http://localhost:8081/](http://localhost:8081/)`​ を実行します。

### [](#キュー内メッセージ集約の-2-番目の例の-xml)キュー内メッセージ集約の 2 番目の例の XML

この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio XML エディターに貼り付けます。

```xml
<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core" xmlns:http="http://www.mulesoft.org/schema/mule/http"
	xmlns:vm="http://www.mulesoft.org/schema/mule/vm"
	xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/vm http://www.mulesoft.org/schema/mule/vm/current/mule-vm.xsd
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
http://www.mulesoft.org/schema/mule/ee/core http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd">
	<http:listener-config name="HTTP_Listener_config" basePath="/" >
		<http:listener-connection host="0.0.0.0" port="8081" />
	</http:listener-config>
	<vm:config name="VM_Config" >
		<vm:connection />
		<vm:queues >
			<vm:queue queueName="demo" />
		</vm:queues>
	</vm:config>
	<flow name="demo-vmFlow1" >
		<http:listener config-ref="HTTP_Listener_config" path="/"/>
		<foreach collection="1 to 55">
			<vm:publish config-ref="VM_Config" queueName="demo" >
				<vm:content ><![CDATA[#["test message"]]]></vm:content>
			</vm:publish>
		</foreach>
	</flow>
	<flow name="demo-vmFlow">
		<scheduler>
			<scheduling-strategy>
				<fixed-frequency frequency="5000" />
			</scheduling-strategy>
		</scheduler>
		<set-variable value="#[[]]" variableName="result" />
		<try>
			<foreach collection="#[1 to 10000]">
				<vm:consume config-ref="VM_Config" queueName="demo" />
				<ee:transform>
					<ee:message>
						<ee:set-payload><![CDATA[%dw 2.0
output application/java
---
vars.result << payload
]]></ee:set-payload>
					</ee:message>
					<ee:variables>
						<ee:set-variable variableName="result"><![CDATA[%dw 2.0
output application/java
---
vars.result<<payload]]></ee:set-variable>
					</ee:variables>
				</ee:transform>
			</foreach>
			<error-handler>
				<on-error-continue enableNotifications="true" logException="false" type="VM:EMPTY_QUEUE" />
			</error-handler>
		</try>
		<set-payload value="#[vars.result]" />
		<foreach collection="#[payload]" counterVariableName="message">
			<logger level="INFO" message="#[payload]" />
		</foreach>
	</flow>
</mule>
```

## [](#関連情報)関連情報

- [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors)
- [MuleSoft ヘルプセンター](https://help.mulesoft.com)