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

Spring ThreadPoolTaskExecutor only running one thread

提问by Jeune

We are using ThreadPoolExecutor in our JMS consumer and injecting it into a DefaultMessageListenerContainer. I expect this to be running concurrent threads for many messages however our logs show that the thread id won't change.Our logging shows that for different processing of messages, the thread id is always the same at 24.

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

This is the spring configuration in that scenario:

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

<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>

After not injecting the threadPoolExectuor bean into the DefaultMessageListenerContainer, the messages are now being executed in different threads.

在没有将 threadPoolExectuor bean 注入 DefaultMessageListenerContainer 之后，这些消息现在正在不同的线程中执行。

This is the resulting configuration:

这是生成的配置：

<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"     
        >   

I have tried reading the documentation and I don't understand why this is happening. Any explanation?

我试过阅读文档，但我不明白为什么会这样。有什么解释吗？

采纳答案by Jeune

After going through the ThreadPoolTaskExecutor code in Springand reading the Java docs for ThreadPoolTaskExecutorI think this is the answer:

在通过Spring 中的ThreadPoolTask​​Executor 代码并阅读ThreadPoolTask​​Executor的 Java 文档后，我认为这就是答案：

Unbounded queues. Using an unbounded queue (for example a LinkedBlockingQueue without a predefined capacity) will cause new tasks to be queued in cases where all corePoolSize threads are busy. Thus, no more than corePoolSize threads will ever be created. (And the value of the maximumPoolSize therefore doesn't have any effect.)

无界队列。在所有 corePoolSize 线程都忙的情况下，使用无界队列（例如，没有预定义容量的 LinkedBlockingQueue）将导致新任务排队。因此，不会创建超过 corePoolSize 的线程。（因此，maximumPoolSize 的值没有任何影响。）

In our configuration above, we were using the LinkedBlockingQueue by default and our corePoolSize is 1. This is why the maximumPoolSize won't have any effect.

在我们上面的配置中，我们默认使用 LinkedBlockingQueue 并且我们的 corePoolSize 是 1。这就是为什么 maximumPoolSize 不会有任何影响。

回答by Jason Chung

change the corePoolSize to 10, then you will get 10 threads run concurrently. Read the javadocon java.util.concurrent.ThreadPoolExecutor which is the backbone of the spring ThreadPoolTaskExecutor, then you will have better understand how to config the corePoolSize and maxPoolSize and queueCapacity

将 corePoolSize 更改为 10，那么您将同时运行 10 个线程。阅读java.util.concurrent.ThreadPoolExecutor 上的javadoc，它是 spring ThreadPoolTask​​Executor 的主干，然后您将更好地了解如何配置 corePoolSize 和 maxPoolSize 以及 queueCapacity

回答by wonhee

I think chosen answer is wrong. IIRC, the way ThreadPoolTaskExecutor(eventually ThreadPoolExecutor in JDK ) working is

我认为选择的答案是错误的。IIRC，ThreadPoolTask​​Executor（JDK 中的最终 ThreadPoolExecutor）的工作方式是

1. ThreadPoolTaskExecutor create threads to up corePoolSize when it's initiated.
2. It takes request up to corePoolSize and let thread to process the task.
3. If there are more requests incoming while all threads are busy, ThreadPoolTaskExecutor is start queuing up those request into internal queue. This can be problematic since this queue size will be Integer.MAX_VALUE as default if you don't specify queue queueCapacity.
4. Request added in #3 will be executed by thread when there is any available threads in the pool.
5. If requests are keep coming and all threads are busy & queue is full, the ThreadPoolTaskExecutor starts creating new threads up to maxPoolSize to process requests.
6. If requests over those (increased threads number + queue size), then task will be rejected or following policy you specified.

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

So the problem here I think is, either 1) your consumer is fast enough or 2) you are stacking requests too slowly, so one thread you specify with corePoolSize was enough to process new incoming requests + queued task without allowing ThreadPoolTaskExecutor to create new threads. I'm pretty sure if you push it harder or set queue's capacity with small number ( like 5~10 ), you'll be able to see the number of threads is increasing.

所以我认为这里的问题是，1）您的消费者足够快，或者 2）您堆叠请求的速度太慢，因此您使用 corePoolSize 指定的一个线程足以处理新的传入请求 + 排队任务，而不允许 ThreadPoolTask​​Executor 创建新线程. 我很确定，如果您加大力度或将队列容量设置为少量（例如 5~10 ），您将能够看到线程数在增加。

回答by Vaibhav

try this:

尝试这个：

<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>

• This will create 10 threads at the time of initialization.
• If all 10 threads are busy and new task comes up, then It will keep tasks in queue.
• If queue is full it will create 11th thread and will go till 25.
• Then will throw TaskRejected Exception.

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