从 tomee 传递到 spring 时忽略 activeMq redeliveryPolicy

Question

我有一个简单的 spring 应用程序部署在 tomee 1.7.1 和 activeMq 5.10 中。 我的问题是我设置的重新投递政策似乎被忽略了，主要是重新投递的延迟。

我是我的 jms 监听器，我立即抛出异常来测试自动重试。

我的 activemq.xml 是这样的：

<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.apache.org/schema/core"
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.xsd
    http://activemq.apache.org/schema/core
    http://activemq.apache.org/schema/core/activemq-core.xsd
">
<bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
<!-- 
    The <broker> element is used to configure the ActiveMQ broker. 
-->
<amq:broker xmlns="http://activemq.apache.org/schema/core" useShutdownHook="false" enableStatistics="true" brokerName="bima" useJmx="true" tmpDataDirectory="${catalina.base}/activemq-data/localhost/tmp_storage" populateJMSXUserID="true" useAuthenticatedPrincipalForJMSXUserID="true" schedulerSupport="true">

    <amq:connectionFactory id="jmsRedeliverConnectionFactory" brokerURL="vm://localhost">
      <amq:redeliveryPolicy>
        <amq:redeliveryPolicy maximumRedeliveries="5" initialRedeliveryDelay="1000" redeliveryUseExponentialBackOff ="true" backOffMultiplier="5" />
      </amq:redeliveryPolicy>

    <amq:persistenceAdapter>
        <amq:kahaDB directory="${catalina.base}/activemq-data/kahadb" checkForCorruptJournalFiles="true" checksumJournalFiles="true" journalMaxFileLength="32mb"/>
    </amq:persistenceAdapter>

    <managementContext>
       <managementContext createConnector="false"/>
    </managementContext>
</amq:broker>

我的 tomee.xml 看起来像这样：

<tomee>
    <Resource id="ActiveMQResourceAdapter" type="ActiveMQResourceAdapter">
        BrokerXmlConfig=broker:(vm://localhost)
        <!-- ServerUrl=vm://localhost -->
    </Resource>

  <!-- see http://tomee.apache.org/containers-and-resources.html -->
  <Resource id="resources/jms/ConnectionFactory" type="javax.jms.ConnectionFactory"> 
       ResourceAdapter = ActiveMQResourceAdapter 
       BrokerURL = vm://localhost
       maximumRedeliveries 3
       redeliveryBackOffMultiplier 2
       redeliveryUseExponentialBackOff true
       initialRedeliveryDelay 5000
       <!-- BrokerUrl = tcp://localhost:61616 -->
    </Resource> 

    <Resource id="resources/jms/XAConnectionFactory" class-name="org.apache.activemq.ActiveMQXAConnectionFactory">
        BrokerURL = vm://localhost

    </Resource>

    <Resource id="resources/jms/PrintQueue" type="javax.jms.Queue"/>


    <Resource id="testxa" class-name="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource">
        Url jdbc:mysql://localhost:3306/test?autoReconnect=true
        User root
    </Resource>

    <Resource id="movieDatabase" type="DataSource">
        XaDataSource testxa
        DataSourceCreator dbcp
        UserName root
    </Resource>

  <!-- activate next line to be able to deploy applications in apps -->
    <Deployments dir="apps" />
</tomee>

我的应用程序上下文中的 jms bean 如下所示：

<bean id="StartQueue" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory">
        <ref local="jmsFactory" />
    </property>
    <property name="defaultDestinationName" value="StartQueue" />
    <property name="deliveryPersistent" value="true"/>
    <property name="explicitQosEnabled" value="true"/>
</bean>
<bean id="starter" class="org.superbiz.mdb.Start"/>
<jms:listener-container container-type="default" connection-factory="jmsFactory" cache="none" acknowledge="auto" transaction-manager="transactionManager"  concurrency="1" >
    <jms:listener destination="StartQueue" ref="starter" />
</jms:listener-container>

一旦我将 maximumRedeliveries 3 添加到开始工作的 tomee xml 中，但重新交付都毫无延迟地发生。所以有些东西忽略了 useExponentialBackOff 和 initialRedeliveryDelay 还是我错过了在某个地方设置它？

编辑：

所以我将 tomee 配置为使用外部代理 activemq 5.10.0 和代理配置

    <!-- START SNIPPET: example -->
<beans
  xmlns="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.xsd
  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">

    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <value>file:${activemq.conf}/credentials.properties</value>
        </property>
    </bean>

    <bean id="logQuery" class="org.fusesource.insight.log.log4j.Log4jLogQuery"
          lazy-init="false" scope="singleton"
          init-method="start" destroy-method="stop">
    </bean>

    <broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" schedulerSupport="true">

        <destinationPolicy>
            <policyMap>
              <policyEntries>
                <policyEntry topic=">" >
                  <pendingMessageLimitStrategy>
                    <constantPendingMessageLimitStrategy limit="1000"/>
                  </pendingMessageLimitStrategy>
                </policyEntry>
              </policyEntries>
            </policyMap>
        </destinationPolicy>

        <plugins>
                    <redeliveryPlugin fallbackToDeadLetter="true" sendToDlqIfMaxRetriesExceeded="true">
                        <redeliveryPolicyMap>
                            <redeliveryPolicyMap>
                                <defaultEntry>
                                    <redeliveryPolicy  maximumRedeliveries="5" initialRedeliveryDelay="1000" redeliveryDelay="1000" useExponentialBackOff ="true"  backOffMultiplier="5"/>
                                </defaultEntry>
                            </redeliveryPolicyMap>
                        </redeliveryPolicyMap>
                    </redeliveryPlugin>
                </plugins>
        <managementContext>
            <managementContext createConnector="false"/>
        </managementContext>

        <persistenceAdapter>
            <kahaDB directory="${activemq.data}/kahadb" checkForCorruptJournalFiles="true" checksumJournalFiles="true" journalMaxFileLength="32mb"/>
        </persistenceAdapter>

          <systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage percentOfJvmHeap="70" />
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="100 gb"/>
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="50 gb"/>
                </tempUsage>
            </systemUsage>
        </systemUsage>

        <transportConnectors>
            <transportConnector name="openwire" uri="tcp://0.0.0.0:61617?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
        </transportConnectors>

        <!-- destroy the spring context on shutdown to stop jetty -->
        <shutdownHooks>
            <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
        </shutdownHooks>

    </broker>
    <import resource="jetty.xml"/>

</beans>
<!-- END SNIPPET: example -->

还有 tomee xml

    <tomee>

    <Resource id="ActiveMQResourceAdapter" type="ActiveMQResourceAdapter">
        # Do not start the embedded ActiveMQ broker
        BrokerXmlConfig  =
        ServerUrl = tcp://10.81.1.28:61617>
    </Resource>

    <Resource id="resources/jms/ConnectionFactory" type="javax.jms.ConnectionFactory"> 
        ResourceAdapter = ActiveMQResourceAdapter 
       maximumRedeliveries 1
    </Resource> 
    <Resource id="testxa" class-name="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource">
        Url jdbc:mysql://localhost:3306/test?autoReconnect=true
        User root
    </Resource>

    <Resource id="movieDatabase" type="DataSource">
        XaDataSource testxa
        DataSourceCreator dbcp
        UserName root
    </Resource>

  <!-- activate next line to be able to deploy applications in apps -->
    <Deployments dir="apps" />
</tomee>

这很有效，当我尝试让 tomee 使用相同的 activemq xml 和 tomee xml 启动 activemq 时，队列会迅速重新传递

    <tomee>
    <Resource id="ActiveMQResourceAdapter" type="ActiveMQResourceAdapter">
        BrokerXmlConfig=broker:(vm://localhost)
        BrokerURL = vm://localhost
    </Resource>

  <Resource id="resources/jms/ConnectionFactory" type="javax.jms.ConnectionFactory">
       ResourceAdapter = ActiveMQResourceAdapter
       maximumRedeliveries 1
    </Resource>


    <Resource id="testxa" class-name="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource">
        Url jdbc:mysql://localhost:3306/test?autoReconnect=true
        User root
    </Resource>

    <Resource id="movieDatabase" type="DataSource">
        XaDataSource testxa
        DataSourceCreator dbcp
        UserName root
    </Resource>

  <!-- activate next line to be able to deploy applications in apps -->
    <Deployments dir="apps" />
</tomee>

配置被忽略并且队列没有被重新交付需要做其他事情以便 tomee 使用重新交付策略？

Answer 1

我认为是 redeliveryUseExponentialBackOff 而不是 useExponentialBackOff：http://activemq.apache.org/maven/5.10.0/apidocs/org/apache/activemq/ra/ActiveMQManagedConnectionFactory.html

基本上使用setter。

Answer 2

为什么不在资源适配器上设置它？

Answer 3

默认情况下，tomee 会停用调度程序服务，因此这可能不适用于嵌入式代理。

您是否尝试过使用外部添加 jar 来支持代理的 xbean xml 配置？ (http://tomee.apache.org/jms-resources-and-mdb-container.html)

Answer 4

回答这个老问题，也许有人会从中受益（我遇到了很多类似的问题）。我遇到了类似的问题，即未正确应用 RedeliveryPolicy。通过将 maximumRedeliveries 设置为 -1，它会无限地重试处理消息，但延迟被忽略了。事实证明，由于 Springs DefaultMessageListenerContainer 上的默认 cacheLevel 字段，重新传递无法正常工作。

将此级别设置为CACHE_CONSUMER 解决了这个问题！