java - DefaultMessageListenerContainer 不缩放

Question

我有一个 DefaultMessageListenerContainer，它(在我看来)没有按比例放大。 Container 被定义为监听一个队列，其中有 100 条消息。

我希望容器可以达到任何长度，消息将尽可能快地被消耗(通过观察 maxConcurrentConsumers 配置)。所以我假设有 7 个并发消费者。 (从容器启动时的 2 个并发消费者开始)一些日志信息:

activeConsumerCount: 5
concurrentConsumers: 2
scheduledConsumerCount: 5
idleConsumerLimit: 1
idleTaskExecLimit: 1
maxConcurrentConsumers: 7

我的 Spring-config(其中一部分):

<bean id="abstractMessageListenerContainer" class="my.package.structure.LoggingListenerContainer" abstract="true">
    <property name="connectionFactory" ref="jmscfCee" />
    <property name="maxConcurrentConsumers" value="7"/>
    <property name="receiveTimeout" value="100000" />
    <property name="concurrentConsumers" value="2" />
</bean>

<bean class="my.package.structure.LoggingListenerContainer" parent="abstractMessageListenerContainer">
    <property name="destinationName" value="MY.QUEUE" />
    <property name="messageListener" ref="myMessageListener" />
</bean>

<bean id="myMessageListener" class="my.package.structure.ListenerClass"></bean>

我的日志容器

public class LoggingListenerContainer extends DefaultMessageListenerContainer{

private static final Logger logger = Logger
        .getLogger(LoggingListenerContainer.class);
@Override
protected void doInvokeListener(MessageListener listener, Message message)
        throws JMSException {

    logger.info("activeConsumerCount: " + this.getActiveConsumerCount());
    logger.info("concurrentConsumers: " +  this.getConcurrentConsumers());
    logger.info("scheduledConsumerCount: " + this.getScheduledConsumerCount());
    logger.info("idleConsumerLimit: " + this.getIdleConsumerLimit());
    logger.info("idleTaskExecLimit: " + this.getIdleTaskExecutionLimit());
    logger.info("maxConcurrentConsumers: " + this.getMaxConcurrentConsumers());
    super.doInvokeListener(listener, message);
}

我的监听类:

public class ListenerClass implements MessageListener {


    public void onMessage(Message msg) {
           //Do some business function
    }

}

有人可以纠正我的配置或给我一些关于我的配置的提示或解释容器的方法吗？ (如果我误解了什么)

我在本地使用 ActiveMQ 进行测试(在使用 WebSphere MQ 的生产环境中)- 如果它与可伸缩性主题相关。

编辑:

<bean id="jmscfCee" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
        <property name="brokerURL">
            <value>${jmscfCee.hostName}</value>
        </property>
</bean>

<bean id="jmscfCeeCachingConnectionFactory"
    class="org.springframework.jms.connection.CachingConnectionFactory ">
    <constructor-arg ref="jmscfCee" />
    <property name="sessionCacheSize" value="10" />
</bean>

Answer 1

这取决于。几年前，我在使用 ActiveMQ 时遇到过类似的问题，它的默认行为针对大量(数千条)小消息进行了大量优化。默认情况下，每个消费者将以 1000 条为一组预取消息，因此如果您的消息数量较少，您可能会发现它们最终都在一个消费者的预取缓冲区中，而其他消费者则闲置。

您可以使用 prefetch policy 调整此行为，或者在连接 URI 上，或者在 Spring 配置中，如果这就是您构建连接工厂的方式。

<amq:connectionFactory id="connectionFactory" brokerURL="vm://localhost">
  <property name="prefetchPolicy">
    <amq:prefetchPolicy all="1" />
  </property>
</amq:connectionFactory>

我当时使用的 ActiveMQ 版本不支持 0 的预取限制(即不预取，每次都去代理)，但文档表明现在允许这样做。