mule和activemq

在本文中，我將展示一個簡單的Mule ESB流程，以了解實際中使用的Active MQ 的DLQ功能 。
我假設您有一個正在運行的Apache ActiveMQ實例（如果沒有，則可以在此處下載一個版本）。 在此示例中，我使用了Mule ESB 3.4.2和ActiveMQ 5.9.0。 我們可以基于以下pom文件創建一個簡單的Mule項目：

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"><modelVersion>4.0.0</modelVersion><groupId>net.pascalalma.demo</groupId><artifactId>activemq-test-flow</artifactId><packaging>mule</packaging><name>${project.artifactId}</name><version>1.0.0-SNAPSHOT</version><properties><mule.version>3.4.2</mule.version><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><jdk.version>1.7</jdk.version><junit.version>4.9</junit.version><activemq.version>5.9.0</activemq.version></properties><dependencies><!-- Mule Dependencies --><dependency><groupId>org.mule</groupId><artifactId>mule-core</artifactId><version>${mule.version}</version></dependency><!-- Mule Transports --><dependency><groupId>org.mule.transports</groupId><artifactId>mule-transport-jms</artifactId><version>${mule.version}</version></dependency><dependency><groupId>org.mule.transports</groupId><artifactId>mule-transport-vm</artifactId><version>${mule.version}</version></dependency><!-- Mule Modules --><dependency><groupId>org.mule.modules</groupId><artifactId>mule-module-client</artifactId><version>${mule.version}</version></dependency><dependency><groupId>org.mule.modules</groupId><artifactId>mule-module-scripting</artifactId><version>${mule.version}</version></dependency><!-- for testing --><dependency><groupId>org.mule.tests</groupId><artifactId>mule-tests-functional</artifactId><version>${mule.version}</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>${junit.version}</version></dependency><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-client</artifactId><version>${activemq.version}</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>2.3.2</version><configuration><source>${jdk.version}</source><target>${jdk.version}</target><encoding>${project.build.sourceEncoding}</encoding></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-resources-plugin</artifactId><version>2.5</version><configuration><encoding>${project.build.sourceEncoding}</encoding></configuration></plugin><plugin><groupId>org.mule.tools</groupId><artifactId>maven-mule-plugin</artifactId><version>1.9</version><extensions>true</extensions><configuration><copyToAppsDirectory>false</copyToAppsDirectory></configuration></plugin></plugins></build> </project>

這里沒有什么特別的。 除了必要的依賴關系之外，我還添加了maven-mule-plugin，以便可以創建“ mule”打包類型并從IDE運行Mule。
使用此Maven pom，我們可以創建以下兩個Mule配置。 一個用于測試交易的Mule流：

<?xml version="1.0" encoding="UTF-8"?> <mule xmlns="http://www.mulesoft.org/schema/mule/core"xmlns:scripting="http://www.mulesoft.org/schema/mule/scripting"version="EE-3.4.1"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.xsdhttp://www.mulesoft.org/schema/mule/scripting http://www.mulesoft.org/schema/mule/scripting/current/mule-scripting.xsd"><flow name="MainFlow"><inbound-endpoint ref="event-queue" /><logger category="net.pascalalma.demo.MainFlow" level="INFO" message="Received message from activeMQ" /><scripting:component><scripting:script engine="Groovy">throw new Exception('Soap Fault Response detected')</scripting:script></scripting:component><outbound-endpoint ref="result-queue" /></flow> </mule>

在此流程中，我們從入站端點接收到一條消息，記錄一條消息并引發異常，然后將該消息放入下一個隊列。 如我們所見，我沒有添加任何異常處理程序。 端點和連接器的配置如下所示：

<?xml version="1.0" encoding="UTF-8"?><mule xmlns="http://www.mulesoft.org/schema/mule/core"xmlns:jms="http://www.mulesoft.org/schema/mule/jms"xmlns:spring="http://www.springframework.org/schema/beans"version="EE-3.4.1"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.xsdhttp://www.mulesoft.org/schema/mule/jms http://www.mulesoft.org/schema/mule/jms/current/mule-jms.xsd"><spring:bean id="redeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy"><spring:property name="maximumRedeliveries" value="5"/><spring:property name="initialRedeliveryDelay" value="500"/><spring:property name="maximumRedeliveryDelay" value="10000"/><spring:property name="useExponentialBackOff" value="false"/><spring:property name="backOffMultiplier" value="3"/></spring:bean><!-- ActiveMQ Connection factory --><spring:bean id="amqFactory" class="org.apache.activemq.ActiveMQConnectionFactory" lazy-init="true"><spring:property name="brokerURL" value="tcp://localhost:61616" /><spring:property name="redeliveryPolicy" ref="redeliveryPolicy" /></spring:bean><jms:activemq-connector name="activeMqConnector"connectionFactory-ref="amqFactory"persistentDelivery="true"numberOfConcurrentTransactedReceivers="2"specification="1.1" /><jms:endpoint name="event-queue" connector-ref="activeMqConnector" queue="event-queue" ><jms:transaction action="ALWAYS_BEGIN" /></jms:endpoint><jms:endpoint name="result-queue" connector-ref="activeMqConnector" queue="result-queue" ><jms:transaction action="ALWAYS_JOIN" /></jms:endpoint> </mule>

我為ActiveMQ連接工廠定義了一個Spring bean，并為該工廠的重新交付策略定義了一個Spring bean。 使用此重新交付策略，我們可以配置當原始嘗試失敗時，Mule應該重試處理隊列中消息的頻率。 重新交付策略中的一個不錯的功能是“ backOffMultiplier”和“ useExponentialBackOff”組合。 使用這些選項，您可以使兩次重新交付嘗試之間的間隔成倍增加，直到達到'maximumRedeliveryDelay'。 在那種情況下，M子將等待“ maximumRedeliveryDelay”等待下一次嘗試。

因此，使用這些配置，我們可以創建一個Mule測試類并運行它。 測試類如下所示：

package net.pascalalma.demo;import org.junit.Test; import org.mule.DefaultMuleMessage; import org.mule.api.MuleMessage; import org.mule.module.client.MuleClient; import org.mule.tck.junit4.FunctionalTestCase;public class TransactionFlowTest extends FunctionalTestCase {@Overrideprotected String getConfigResources() {return "app/test-flow.xml, app/test-endpoints.xml";}@Testpublic void testError() throws Exception {MuleClient client = new MuleClient(muleContext);MuleMessage inMsg = new DefaultMuleMessage("<txt>Some message</txt>", muleContext);client.dispatch("event-queue", inMsg);// Give Mule the chance to redeliver the messageThread.sleep(4000);} }

如果我們運行此測試，您將在日志中看到如下消息：

Exception stack is: 1. "Message with id "ID:Pascals-MacBook-Pro-2.local-59158-1406440948059-1:1:3:1:1" has been redelivered 3 times on endpoint "jms://event-queue", which exceeds the maxRedelivery setting of 0 on the connector "activeMqConnector". Message payload is of type: ActiveMQTextMessage (org.mule.transport.jms.redelivery.MessageRedeliveredException)org.mule.transport.jms.redelivery.JmsXRedeliveryHandler:87 (http://www.mulesoft.org/docs/site/current3/apidocs/org/mule/transport/jms/redelivery/MessageRedeliveredException.html)

如果現在切換到ActiveMQ控制臺 ，可以通過http：// localhost：8161進行默認本地安裝，則可以看到以下隊列：

不出所料，我們看到正在創建兩個隊列，事件隊列為空，默認的ActiveMQ.DLQ包含我們的消息：

如您所見，為每個隊列指定一個特定的DLQ而不是一個將包含所有無法傳遞的消息的DLQ可能很方便。 幸運的是，這很容易在ActiveMQ中進行配置。 只需將以下內容放入“ $ ACTIVEMQ_HOME / conf”文件夾中的“ activemq.xml”文件中。

<!-- Set the following policy on all queues using the '>' wildcard --> <policyEntry queue=">"><deadLetterStrategy><individualDeadLetterStrategy queuePrefix="DLQ." useQueueForQueueMessages="true" /></deadLetterStrategy> </policyEntry>

如果現在重新啟動ActiveMQ，請刪除現有隊列并重新運行測試，我們將看到以下結果：

因此，使用此設置，每個隊列都有自己的DLQ。 有關這些ActieMQ設置的更多選項，請參見此處 。 通過本文中創建的Mule流，可以輕松測試和使用這些設置。

翻譯自: https://www.javacodegeeks.com/2014/07/mule-esb-activemq-and-the-dlq.html

mule和activemq

總結

以上是生活随笔為你收集整理的mule和activemq_Mule ESB，ActiveMQ和DLQ的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。