我刚开始使用JMS/ActiveMQ,我有一个Spring/Hibernate应用程序,可以从ActiveMQ的队列中获取消息并处理这些消息以进行持久化。由于消息需要一段时间来处理和持久化,因此我将我的DefaultMessageListenerContainer配置为具有多个消费者(例如5-10),以便可以同时处理多个消息。我查看了很多ActiveMQ和Spring API文档,我认为我所需要做的就是将maxConcurrentConsumers设置为10+concurrentConsumers设置为5,或者在DefaultMessageListenerContainer上将并发设置为5-10。一旦我这样做了,我可以从ActiveMQ的内置控制台中看到我的队列确实有5个消费者。但是,当我将10或100条消息放入队列时,处理似乎是单线程的,并且我添加了一行日志来打印线程ID,它似乎是同一个线程ID按顺序处理所有请求。从控制台的ActiveMQ队列页面上,我点击浏览活动消费者链接来查看正在发生什么,看起来一个消费者有100个待处理的消息,其他4个没有任何消息。
我做了一些研究,找到了这篇Spring的文章(http://forum.springsource.org/showthread.php?61170-Messages-missed-using-DefaultMessageListenerContainer),并添加了一个预取策略,值为2,认为每个消费者都会签收1000条消息。现在当我发送另一批消息时,一个消费者将有2-3条消息挂起,但其他4个消费者仍然空闲,最终由那个消费者顺序处理所有内容。此时我想可能是我在ActiveMQ代理上进行的错误配置。我在文档中读到默认的分派策略是循环策略,但我看到了一个名为constantPendingMessageLimitStrategy的设置,在我的activemq.xml中设置为1000,并尝试将其设置为非常低的数字(例如2),以为它控制代理一次发送给消费者多少条消息,但仍然没有任何作用。希望有人能指出我做错了什么,我已经发布了我的Spring配置,除了尝试那个设置(constantPendingMessageLimitStrategy)之外,我真的没有触及activemq.xml文件。我正在使用ActiveMQ 5.8。
<bean id="importRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
<property name="initialRedeliveryDelay" value="15000" />
<property name="maximumRedeliveries" value="-1" />
<property name="useExponentialBackOff" value="true" />
<property name="backOffMultiplier" value="2" />
</bean>
<bean id="importPrefetchPolicy" class="org.apache.activemq.ActiveMQPrefetchPolicy">
<property name="all" value="2"></property>
</bean>
<bean id="importConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="${import.queue.url}"/>
<property name="redeliveryPolicy" ref="importRedeliveryPolicy" />
<property name="prefetchPolicy" ref="importPrefetchPolicy"></property>
</bean>
<bean id="importQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="${import.queue.name}" />
</bean>
<bean id="importListener" class="com.mycompany.ImportQueueListener" >
<property name="importService" ref="importService"></property>
<property name="sessionFactory" ref="sessionFactory"/>
</bean>
<bean id="importJmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="importConnectionFactory" />
<property name="destination" ref="importQueue" />
<property name="messageListener" ref="importListener" />
<property name="sessionTransacted" value="true" />
<property name="maxConcurrentConsumers" value="10"></property>
<property name="concurrentConsumers" value="5"></property>
</bean>