VM Connector を使用したキュー内保留メッセージの集約の例
次の例は、仮想マシン用 Anypoint Connector (VM Connector) を使用して、キューで保留中のメッセージを集約する方法についての 2 つのシナリオを示しています。
VM はキューであり、キュー内で保留中のすべてのメッセージに対して for each 操作を実行することはできません。
キュー内メッセージ集計の最初の例
この例では、Aggregators Module を使用して、VM Listener 操作によって受信されたメッセージを集約し、集約の結果を反復処理します。
最初のフローの作成
最初のフローでは、HTTP [Listener] ソースが受信 HTTP メッセージをリスンします。次に、VM Publish 操作が VM キュー内の特定のコンテンツをパブリッシュします。
-
Studio で、新しい Mule プロジェクトを作成します。
-
[Mule Palette (Mule パレット)] で、[HTTP] > [Listener] ソースを選択してキャンバスにドラッグします。
-
HTTP [Listener] 設定画面で、[Path (パス)] 項目を
/ に設定します。 -
[Connector configuration (コネクタ設定)] 項目の横にあるプラス記号 (+) をクリックして、アプリケーション内の HTTP Listener のすべてのインスタンスで使用できるグローバル要素を設定します。
-
HTTP グローバル設定の必須項目を入力します。
-
[OK] をクリックします。
-
[For Each] コンポーネントを HTTP の [Listener] ソースの右にドラッグします。
-
[Collection (コレクション)] を
1 to 200 に設定します。 -
[VM] > [Publish] 操作を [For Each] コンポーネントにドラッグします。
-
[Content (コンテンツ)] を
"hello world" に設定します。 -
[Connector configuration (コネクタ設定)] 項目の横にあるプラス記号 (+) をクリックして、アプリケーション内の VM Publish 操作のすべてのインスタンスで使用できるグローバル要素を設定します。
-
[VM Config (VM 設定)] ウィンドウで、プラス記号 (+) をクリックして新しいキューを追加します。
-
[Queue name (キュー名)] を
demo に設定します。 -
[Finish (完了)] をクリックします。
-
[OK] をクリックします。
2 つ目のフローの作成
2 番目のフローでは VM [Listener] 操作が VM キューにパブリッシュされたメッセージをリスンします。[Time based aggregator (時間ベースのアグリゲーター)] スコープを使用すると、時間条件を満たすまで受信したメッセージを集約することができます。
-
Studio で [VM] > [Listener] 操作を最初のフローの下にドラッグします。
-
[Connector configuration (コネクタ設定)] で、前に作成した
VM_Config 設定を選択します。 -
[Queue name (キュー名)] を
demo に設定します。 -
[Time based aggregator (時間ベースのアグリゲーター)] スコープを [VM Listener] の右にドラッグします。
-
[Name (名前)] を
demoaggregator に設定します。 -
[Period (期間)] を
5 に設定して、5 秒間に達するまで受信メッセージを集約します。
3 つ目のフローを作成する
このフローでは、[Aggregator listener] 操作は [Time based aggregator (時間ベースのアグリゲーター)] スコープによって集約されたメッセージをリスンします。[For Each] コンポーネントはメッセージを反復処理し、[Logger] コンポーネントは各メッセージのコンテンツを記録します。
-
Studio で、[Aggregator listener] ソースを 2 番目のフローの上にドラッグします。
-
[Aggregator name (アグリゲーター名)] を
demoaggregator に設定します。 -
[For Each] コンポーネントを [Aggregator listener] ソースの右にドラッグします。
-
[Collection (コレクション)] を
[#payload] に設定します。 -
[Logger] コンポーネントを [For Each] コンポーネントにドラッグします。
-
[Message (メッセージ)] を
[#payload] に設定します。 -
Mule アプリケーションを保存して実行します。
-
ターミナルで、curl コマンド
curl http://localhost:8081/ を実行します。
キュー内メッセージ集約の最初の例の XML
この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio 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>xmlキュー内メッセージ集計の 2 番目の例
この例では、[For Each] コンポーネントは VM [Consume] 操作によってコンシュームされたメッセージを反復処理します。Mule アプリケーションは受信したすべてのメッセージを変数に追加し、各メッセージのペイロードコンテンツを記録します。この例のシナリオではキューにメッセージがなくなるまでメッセージを取得します。キューにメッセージがなくなると、VM [Consume] 操作はタイムアウトメッセージをスローし、メッセージは [On Error Continue] コンポーネントで処理され、結果が使用可能になります。
最初のフローの作成
最初のフローでは [HTTP Listener] ソースが受信 HTTP メッセージをリスンし、VM [Publish] 操作が VM キューに特定のコンテンツをパブリッシュします。
-
Studio で、新しい Mule プロジェクトを作成します。
-
[Mule Palette (Mule パレット)] で、[HTTP] > [Listener] ソースを選択してキャンバスにドラッグします。
-
HTTP [Listener] 設定画面で、[Path (パス)] 項目を
/ に設定します。 -
[Connector configuration (コネクタ設定)] 項目の横にあるプラス記号 (+) をクリックして、アプリケーション内の HTTP Listener のすべてのインスタンスで使用できるグローバル要素を設定します。
-
HTTP グローバル設定の必須項目を入力します。
-
[OK] をクリックします。
-
[For Each] コンポーネントを HTTP の [Listener] ソースの右にドラッグします。
-
[Collection (コレクション)] を
1 to 55 に設定します。 -
[VM] > [Publish] 操作を [For Each] コンポーネントにドラッグします。
-
[Content (コンテンツ)] を
"test world" に設定します。 -
[Connector configuration (コネクタ設定)] 項目の横にあるプラス記号 (+) をクリックして、アプリケーション内の VM Publish 操作のすべてのインスタンスで使用できるグローバル要素を設定します。
-
[VM Config (VM 設定)] ウィンドウで、プラス記号 (+) をクリックして新しいキューを追加します。
-
[Queue name (キュー名)] を
demo に設定します。 -
[Finish (完了)] をクリックします。
-
[OK] をクリックします。
2 つ目のフローの作成
2 番目のフローでは VM [Consume] 操作がキューからパブリッシュされたメッセージを取得します。[Transform Message] コンポーネントは、ペイロードコンテンツ変換し、変数に保存します。Mule アプリケーションは各変数ペイロードを記録します。
-
Studio で [Scheduler] ソースを最初のフローの下にドラッグします。
-
[Frequency (頻度)] を
5000 に設定します。 -
[Set Variable] コンポーネントを [Scheduler] コンポーネントの右にドラッグします。
-
[Name (名前)] を
result に設定します。 -
[Value (値)] を
[] に設定します。 -
[Try] スコープコンポーネントを [Set Variable] の右にドラッグします。
-
[For Each] コンポーネントを [Try] スコープにドラッグします。
-
[Collection (コレクション)] を
1 to 10000 に設定します。 -
VM [Consume] 操作を [For Each] コンポーネントにドラッグします。
-
[Connector configuration (コネクタ設定)] で、前に作成した
VM_Config 設定を選択します。 -
[Queue name (キュー名)] を
demo に設定します。 -
[Transform Message] コンポーネントを [Consume] の右にドラッグします。
-
DataWeave スクリプトを次のように設定します。
%dw 2.0 output application/java --- vars.result << payload
-
[On Error Continue] コンポーネントを [Try] スコープコンポーネントの [Error handling (エラー処理)] セクションにドラッグします。
-
[Type (種別)] を
VM:EMPTY_QUEUE に設定します。 -
[Set Payload] コンポーネントを [Try] スコープコンポーネントの右にドラッグします。
-
[Value (値)] を
vars.result に設定します。 -
[For Each] コンポーネントを [Set Payload] コンポーネントの右にドラッグします。
-
[Collection (コレクション)] を
payload に設定します。 -
[Logger] コンポーネントを [For Each] スコープコンポーネントにドラッグします。
-
[Message (メッセージ)] を
payload に設定します。 -
Mule アプリケーションを保存して実行します。
-
ターミナルで、curl コマンド
curl http://localhost:8081/ を実行します。
キュー内メッセージ集約の 2 番目の例の XML
この例のフローをすばやく Mule アプリケーションに読み込むには、次のコードを Studio 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>xml