Spring JMS侦听器-容器并发属性不工作

Question

大家好，我正在用ActiveMQ学习Spring JMS。在我的示例场景中，生产者应用程序在队列中发送大约50条消息，当我启动消费者应用程序时，它开始使用这些消息。

为每个侦听器启动的并发会话/使用者的数量。可以是表示最大值的简单数字（例如“5”），也可以是表示下限和上限的范围（例如“3-5”）。请注意，指定的最小值只是一个提示，可能会在运行时被忽略。默认值为1；如果有主题侦听器或队列排序很重要,则将并发限制为1；请考虑为一般队列引发它。

但在我的配置中，我将该属性设置为5，但似乎无法启动5个并发侦听器。

侦听器的配置：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jms="http://www.springframework.org/schema/jms"

    xmlns:p="http://www.springframework.org/schema/p"
    xsi:schemaLocation="http://www.springframework.org/schema/beans 
    http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
    http://www.springframework.org/schema/jms
    http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">

    <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" p:brokerURL="tcp://localhost:61616" />

    <bean id="listener" class="com.jms.example.MyMessageListener"></bean>

    <jms:listener-container container-type="default" concurrency="5"
        connection-factory="connectionFactory">
        <jms:listener destination="MyQueue" ref="listener"
            method="onMessage"></jms:listener>
    </jms:listener-container>

</beans>

<bean id="msgListenerContainer" 
        class="org.springframework.jms.listener.DefaultMessageListenerContainer"
        p:connectionFactory-ref="connectionFactory"
        p:destination-ref="destination"
        p:messageListener-ref="listener"
        p:concurrentConsumers="10"
        p:maxConcurrentConsumers="50" />

编辑：

消费者代码：

public class MyMessageListener implements MessageListener{


    public void onMessage(Message m) {
        TextMessage message=(TextMessage)m;
        try{
            System.out.println("Start = " + message.getText());
            Thread.sleep(5000);
            System.out.println("End = " + message.getText());
        }catch (Exception e) {e.printStackTrace();  }
    }
}

我正在控制台上打印已消耗的消息,其输出将在下面的方案中解释：

null

这里的问题是它没有加载所有的10个消费者。有时它装载3或1。

Start = hello jms 1 // consumer 1 started 
Start = hello jms 2 // consumer 2 started 
Start = hello jms 3 // consumer 3 started 
End = hello jms 1  //  consumer 1 ended
Start = hello jms 4 // consumer 4 started and hence always 3 consumers and not 10
End = hello jms 2
Start = hello jms 5
End = hello jms 3
Start = hello jms 6

情景2：

1. 启动生产者并发送消息（同时使用者正在运行）
2. 由于使用者处于运行状态，它将开始使用它们。

Start = hello jms 1 // consumer 1 started 
Start = hello jms 2 // consumer 2 started 
Start = hello jms 3 // consumer 3 started 
Start = hello jms 4 // consumer 4 started 
Start = hello jms 5 // consumer 5 started 
Start = hello jms 6 // consumer 6 started 
Start = hello jms 7 // consumer 7 started 
Start = hello jms 8 // consumer 8 started 
Start = hello jms 9 // consumer 9 started 
Start = hello jms 10 // consumer 10 started. Hence all them started at same time as expected.
End = hello jms 1
Start = hello jms 11
End = hello jms 2
Start = hello jms 12
End = hello jms 3
Start = hello jms 13

请帮帮忙。

Answer 1

就像Strelok告诉我的关于预取信息的事情一样。创建了PrefetchPolicybean，其中QueuePrefetch属性设置为1。其引用在ConnectionFactory中设置。

我在配置上做了一些改动，这些改动如下：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jms="http://www.springframework.org/schema/jms"

    xmlns:p="http://www.springframework.org/schema/p"
    xsi:schemaLocation="http://www.springframework.org/schema/beans 
    http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
    http://www.springframework.org/schema/jms
    http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">

    <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" p:brokerURL="tcp://localhost:61616"
        p:prefetchPolicy-ref="prefetchPolicy" />

    <bean id="prefetchPolicy" class="org.apache.activemq.ActiveMQPrefetchPolicy"
        p:queuePrefetch="1" />

    <bean id="listener" class="com.javatpoint.MyMessageListener"></bean>

    <jms:listener-container concurrency="10-15" connection-factory="connectionFactory">
        <jms:listener destination="javatpointQueue" ref="listener"
            method="onMessage"></jms:listener>
    </jms:listener-container>

    <!-- The JMS destination -->

      <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="javatpointQueue" />
      </bean>
</beans>