Mule Persistent ActiveMQ RedeliveryPolicy

Question

我正在使用Mule作爲ESB解決方案。我有一個隊列，從那裏我收到消息，並試圖使HTTP請求服務，始終失敗。

我已經配置了RedeliveryPolicy上的ActiveMQ這樣的：

<spring:bean id="retryRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy" 
     name="retryRedeliveryPolicy"> 
     <spring:property name="maximumRedeliveries" value="76" /> 
     <spring:property name="initialRedeliveryDelay" value="300000" /> 
     <spring:property name="maximumRedeliveryDelay" value="3600000" /> 
     <spring:property name="useExponentialBackOff" value="true" /> 
     <spring:property name="backOffMultiplier" value="2" /> 
     <spring:property name="queue" value="*" /> 
    </spring:bean> 

這5分鐘後重試。然後10分鐘，20,40,60,60,60 ...約~3天

問題是，重試邏輯是非持久性。

可以說，郵件正在重試2天。而且我已經部署了新版本的mule應用程序或重新啓動的服務器......在這種情況下，重試邏輯將從5分鐘，10分鐘重新開始......因爲重試狀態由客戶端保存在RAM內存中。

很熱做RedeliveryPolicy persistent？在2天后我將重新啓動服務器後，它必須繼續重試1天。

一個解決方案，我認爲可能是設置timeToLive 72小時的消息。但即使如此，重新啓動服務器後。它不會從開始每小時重試。它將從5分鐘開始...

Answer 1

您可以使用JMS消息的JMSXDeliveryCount屬性來檢查它已被重試了多少次。重試時間間隔邏輯應該依賴於JMSXDeliveryCount變量。

message.getIntProperty("JMSXDeliveryCount ")

http://activemq.apache.org/activemq-message-properties.html

Answer 2

這是沒有辦法讓RedeliveryPolicy持久 - 它是由連接工廠控制，當你重新啓動服務器的工廠將被重置。你配置它的方式，它會繼續你看到的行爲，因爲你有useExponentialBackOff設置爲true。您可以將此設置爲false以使延遲成爲常規數量，但這是唯一需要做出的更改。

我認爲你有正確的想法設置消息TTL，至少這將消除指定時間後的消息。 JMSXDeliveryCount非常好，但是如果增加重試之間的延遲，它會在數次嘗試後刪除，而不是定義的時間段。

Answer 3

ActiveMQ有一種方法可以執行持久性的重新傳遞，但它不是在ConnectionFactory中使用RedeliveryPolicy構建的，它在失敗前用於短時間的重新傳遞。

但是，可以使用ActiveMQ調度程序和延遲的消息來構建持久的重新傳送。有點手動，但可行。確保在嘗試此操作之前在ActiveMQ中打開schedulerSupport =「true」。

JMS屬性：「delivery」記錄重試次數，如果出現問題，catch異常策略會延遲多次重新發送消息。實際的延遲由ActiveMQ處理，所以這個流程可以處理數小時，數天等的延遲，並承受ActiveMQ和Mule的重啓。

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

<mule xmlns:scripting="http://www.mulesoft.org/schema/mule/scripting" 
    xmlns:jms="http://www.mulesoft.org/schema/mule/jms" xmlns:metadata="http://www.mulesoft.org/schema/mule/metadata" 
    xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation" 
    xmlns:spring="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-current.xsd 
http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd 
http://www.mulesoft.org/schema/mule/jms http://www.mulesoft.org/schema/mule/jms/current/mule-jms.xsd 
http://www.mulesoft.org/schema/mule/scripting http://www.mulesoft.org/schema/mule/scripting/current/mule-scripting.xsd"> 

    <spring:beans> 
     <spring:bean name="redeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy"> 
      <spring:property name="maximumRedeliveries" value="0" /> 
     </spring:bean> 

     <spring:bean name="cf" 
      class="org.apache.activemq.ActiveMQConnectionFactory"> 
      <spring:property name="brokerURL" value="tcp://localhost:61616" /> 
      <spring:property name="redeliveryPolicy" ref="redeliveryPolicy" /> 
     </spring:bean> 
    </spring:beans> 

    <jms:activemq-connector name="Active_MQ" 
     specification="1.1" brokerURL="tcp://localhost:61616" 
     validateConnections="true" doc:name="Active MQ" numberOfConsumers="1" 
     maxRedelivery="-1" connectionFactory-ref="cf" /> 

    <flow name="testFlow"> 
     <jms:inbound-endpoint queue="IN" connector-ref="Active_MQ" 
      doc:name="JMS"> 
      <jms:transaction action="ALWAYS_BEGIN" /> 
     </jms:inbound-endpoint> 

     <set-property propertyName="delivery" 
      value="#[message.inboundProperties['delivery'] or 0]" doc:name="Property" /> 

     <scripting:component doc:name="Throw Error"> 
      <scripting:script engine="Groovy"><![CDATA[throw new java.lang.RuntimeException("Error in processing")]]></scripting:script> 
     </scripting:component> 

     <choice-exception-strategy doc:name="Choice Exception Strategy"> 
      <catch-exception-strategy doc:name="Catch Exception Strategy" 
       when="#[message.outboundProperties['delivery'] &lt; 5]"> 
       <logger level="ERROR" 
        message="Retry once more count #[message.outboundProperties['delivery']]" 
        doc:name="Logger" /> 

       <set-property propertyName="AMQ_SCHEDULED_DELAY" value="3000" 
        doc:name="Property" /> 
       <set-property propertyName="delivery" 
        value="#[message.outboundProperties['delivery'] + 1]" doc:name="Property" /> 
       <jms:outbound-endpoint queue="IN" 
        connector-ref="Active_MQ" doc:name="JMS"> 
        <jms:transaction action="ALWAYS_JOIN" /> 
       </jms:outbound-endpoint> 

      </catch-exception-strategy> 

      <rollback-exception-strategy doc:name="Rollback Exception Strategy"> 
       <logger level="ERROR" message="Giving up retry. Do whatever needed here." doc:name="Logger" /> 
      </rollback-exception-strategy> 
     </choice-exception-strategy> 
    </flow> 
</mule>