Spring ThreadPoolTaskExecutor работает только один поток

Question

Мы используем ThreadPoolExecutor в нашем клиенте JMS и внедряем его в DefaultMessageListenerContainer. Я ожидаю, что во многих сообщениях будут запущены параллельные потоки, однако наши журналы показывают, что идентификатор потока не изменится. Наши журналы показывают, что для различной обработки сообщений идентификатор потока всегда одинаков при 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 не внедрен в 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

После прохождения кода ThreadPoolTaskExecutor в Spring и чтения документов Java для ThreadPoolTaskExecutor Я думаю, что это ответ:

Неограниченные очереди. Использование неограниченного очередь (например, LinkedBlockingQueue без предопределенная емкость) вызовет новый задачи в очереди в случаях, когда все Потоки corePoolSize заняты. Таким образом, не больше чем corePoolSize темы будут когда-либо быть созданным. (И значение MaximumPoolSize поэтому не имеет любой эффект.)

В нашей конфигурации выше, мы использовали LinkedBlockingQueue по умолчанию, и наш corePoolSize равен 1. Вот почему MaximumPoolSize не будет иметь никакого эффекта.

Answer 3

Я думаю, что выбранный ответ неверен. IIRC, способ ThreadPoolTaskExecutor (в конце концов ThreadPoolExecutor в JDK) работает

1. ThreadPoolTaskExecutor создает потоки для обновления corePoolSize при его запуске.
2. Он принимает запрос до corePoolSize и позволяет потоку обрабатывать задачу.
3. Если поступает больше запросов, когда все потоки заняты, ThreadPoolTaskExecutor начинает помещать эти запросы во внутреннюю очередь. Это может быть проблематично, так как этот размер очереди будет Integer.MAX_VALUE по умолчанию, если вы не укажете queueCueCapacity.
4. Запрос, добавленный в # 3, будет выполняться потоком, когда в пуле есть все доступные потоки.
5. Если запросы продолжают поступать, и все потоки заняты, а очередь заполнена, ThreadPoolTaskExecutor начинает создавать новые потоки вплоть до maxPoolSize для обработки запросов.
6. Если запросы превышают эти (увеличение числа потоков + размер очереди), тогда задача будет отклонена или будет соответствовать указанной вами политике.

Так что проблема здесь, я думаю, заключается либо в том, что либо 1) ваш потребитель достаточно быстр, либо 2) вы слишком медленно складываете запросы, поэтому одного потока, указанного с помощью corePoolSize, было достаточно для обработки новых входящих запросов + задача в очереди, не позволяя ThreadPoolTaskExecutor создавать новые темы. Я почти уверен, что если вы нажмете его сильнее или установите емкость очереди с небольшим числом (например, 5 ~ 10), вы увидите, что количество потоков увеличивается.

Answer 4

измените corePoolSize на 10, тогда вы получите одновременный запуск 10 потоков. Прочтите javadoc на java.util.concurrent.ThreadPoolExecutor, который является основой пружины ThreadPoolTaskExecutor, тогда вы лучше поймете, как настроить corePoolSize и maxPoolSize и queueCapacity