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

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

Это конфигурация spring в этом сценарии:

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

Я пробовал прочитать документацию, и я не понимаю, почему это происходит. Любое объяснение?

Ответ 1

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

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

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

Ответ 2

попробуйте следующее:

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

Ответ 3

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

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

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

Ответ 4

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