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

<?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>

這里沒有什么特別的。 除了必要的依賴關(guān)系之外，我還添加了maven-mule-plugin，以便可以創(chuàng)建“ mule”打包類型并從IDE運(yùn)行Mule。
有了此Maven pom，我們可以創(chuàng)建以下兩個(gè)Mule配置。 一個(gè)用于測試交易的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>

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

<?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連接工廠定義了一個(gè)Spring bean，并為該工廠的重新交付策略定義了一個(gè)Spring bean。 使用此重新分發(fā)策略，我們可以配置當(dāng)原始嘗試失敗時(shí)，Mule應(yīng)該重試處理隊(duì)列中消息的頻率。 重新交付策略中的一個(gè)不錯(cuò)的功能是“ backOffMultiplier”和“ useExponentialBackOff”組合。 使用這些選項(xiàng)，您可以使兩次重新交付嘗試之間的間隔成倍增加，直到達(dá)到'maximumRedeliveryDelay'。 在那種情況下，M子將等待“ maximumRedeliveryDelay”等待下一次嘗試。

因此，使用這些配置，我們可以創(chuàng)建一個(gè)Mule測試類并運(yùn)行它。 測試類如下所示：

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);} }

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

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)

如果現(xiàn)在我們切換到ActiveMQ控制臺(tái) ，可以通過http：// localhost：8161進(jìn)行默認(rèn)本地安裝，則可以看到以下隊(duì)列：
不出所料，我們看到正在創(chuàng)建兩個(gè)隊(duì)列，事件隊(duì)列為空，默認(rèn)的ActiveMQ.DLQ包含我們的消息：

正如您可以想象的那樣，為每個(gè)隊(duì)列使用一個(gè)特定的DLQ而不是一個(gè)包含各種無法傳遞的消息的DLQ可能很方便。 幸運(yùn)的是，這很容易在ActiveMQ中配置。 只需將以下內(nèi)容放在“ $ 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>

如果現(xiàn)在重新啟動(dòng)ActiveMQ，請刪除現(xiàn)有隊(duì)列并重新運(yùn)行測試，我們將看到以下結(jié)果：
因此，使用此設(shè)置，每個(gè)隊(duì)列都有自己的DLQ。 有關(guān)這些ActieMQ設(shè)置的更多選項(xiàng)，請參見此處 。 通過本文中創(chuàng)建的Mule流，可以輕松測試和使用這些設(shè)置。

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。