Spring ThreadPoolTask​​Executor只运行一个线程

Question

我们在JMS使用者中使用ThreadPoolExecutor并将其注入DefaultMessageListenerContainer.我希望这会为许多消息运行并发线程,但是我们的日志显示线程id不会改变.我们的日志记录显示,对于不同的消息处理,线程ID在24处始终相同.

这是该场景中的弹簧配置:

<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer"       
         p:connectionFactory-ref="cachedConnectionFactory"
         p:destination-ref="formsCRRDestination"
         p:messageListener-ref="formServicePojo"
         p:concurrentConsumers="5"
         p:idleTaskExecutionLimit="1"
         p:maxConcurrentConsumers="25"
         p:taskExecutor-ref="threadPoolExecutor"         
         destroy-method="doShutdown"     
    >   


 <bean id="threadPoolExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" >
        <property name="corePoolSize" value="1"/>
        <property name="maxPoolSize" value="15"/>
        <property name="keepAliveSeconds" value="30"/>
    </bean>

在不将threadPoolExectuor bean注入DefaultMessageListenerContainer之后,消息现在正在不同的线程中执行.

这是最终的配置:

<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer"       
             p:connectionFactory-ref="cachedConnectionFactory"
             p:destination-ref="formsCRRDestination"
             p:messageListener-ref="formServicePojo"
             p:concurrentConsumers="5"
             p:idleTaskExecutionLimit="1"
             p:maxConcurrentConsumers="25"       
             destroy-method="doShutdown"     
        >   

我试过阅读文档,但我不明白为什么会这样.任何解释？

Answer 1

试试这个:

<bean id="threadPoolTaskExecutor"
        class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
        <property name="corePoolSize" value="10" />
        <property name="maxPoolSize" value="25" />
        <property name="queueCapacity" value="30" />
</bean>

• 这将在初始化时创建10个线程.
• 如果所有10个线程都忙,并且新任务出现,那么它将把任务保留在队列中.
• 如果队列已满,它将创建第11个线程,并将持续到25.
• 然后会抛出TaskRejected Exception.

Answer 2

在浏览Spring中的ThreadPoolTask​​Executor代码并阅读ThreadPoolTask​​Executor的Java文档后,我认为这就是答案:

无限的队列.使用无界队列(例如,没有预定义容量的LinkedBlockingQueue)将导致在所有corePoolSize线程忙的情况下将新任务排队.因此,只会创建corePoolSize线程.(而且maximumPoolSize的值因此没有任何影响.)

在上面的配置中,我们默认使用LinkedBlockingQueue,我们的corePoolSize为1.这就是maximumPoolSize不会产生任何影响的原因.

Answer 3

我认为选择的答案是错误的.IIRC,ThreadPoolTask​​Executor(最终是JDK中的ThreadPoolExecutor)的工作方式

1. ThreadPoolTask​​Executor在启动时创建线程到corePoolSize.
2. 它接受corePoolSize的请求并让线程处理任务.
3. 如果在所有线程忙的情况下有更多请求传入,则ThreadPoolTask​​Executor开始将这些请求排队到内部队列中.这可能会有问题,因为如果您未指定queue queueCapacity,则此队列大小将默认为Integer.MAX_VALUE.
4. 当池中有任何可用线程时,线程将执行#3中添加的请求.
5. 如果请求继续发生并且所有线程都忙并且队列已满,则ThreadPoolTask​​Executor开始创建新线程,直到maxPoolSize来处理请求.
6. 如果请求超过这些请求(增加的线程数+队列大小),则任务将被拒绝或遵循您指定的策略.

所以我认为这里的问题是,1)您的消费者足够快或2)您堆叠请求太慢,因此您使用corePoolSize指定的一个线程足以处理新的传入请求+排队任务而不允许ThreadPoolTask​​Executor创建新线程.我很确定如果你更努力地推动它或者用小数字设置队列的容量(比如5~10),你将能够看到线程的数量正在增加.