Настройте интеграцию Spring. ErrorHandlingTaskExecutor

Question

<context:annotation-config/>
<context:component-scan base-package="ru.*"/>

<int:poller id="defaultPoller" default="true"
            fixed-delay="1000" task-executor="taskExecutor"/>
<task:executor id="taskExecutor" pool-size="1-5" queue-capacity="200" rejection-policy="CALLER_RUNS"/>

<int:channel id="listFiles">
    <int:queue capacity="1000"/>
</int:channel>

<int:channel id="pages">
    <int:queue capacity="1000"/>
</int:channel>

<int:channel id="resultWithHeader">
    <int:queue capacity="1000"/>
</int:channel>

<int:channel id="toBD">
    <int:queue capacity="1000"/>
</int:channel>

<int:channel id="unparsedToUniq">
    <int:queue capacity="1000"/>
</int:channel>

<task:scheduled-tasks>
    <task:scheduled ref="getFilesList"
                    method="getList"
                    cron="0 0 12 * * WED"/>
</task:scheduled-tasks>

<int:splitter id="splitterPdf" ref="split" method="receive" input-channel="listFiles" output-channel="pages"/>

<int:transformer id="transfMain" input-channel="pages" ref="firstTransform" method="transform"
                 output-channel="resultWithHeader"/>

<int:header-value-router id="routedParsingDocument" input-channel="resultWithHeader" header-name="parsed">
    <int:mapping value="yes" channel="toBD"/>
    <int:mapping value="no" channel="unparsedToUniq"/>
</int:header-value-router>

<int:transformer id="transfUniq" input-channel="unparsedToUniq" ref="secondTransform" method="transform"
                 output-channel="toBD"/>

<int:service-activator id="bd_activator" input-channel="toBD" method="receive" ref="bd" > <int:poller task-executor="taskBD" fixed-delay="1500"/></int:service-activator>

<bean id="getFilesList" class="ru.*.GetAndReadFiles"/>

<bean id="split" class="ru.*.SplitDocument"/>

<bean id="firstTransform" class="ru.*.MainParser"/>

<bean id="secondTransform" class="ru.*.UniqParser"/>

<bean id="bd" class="ru.*.BDWriter"/>

<task:executor id="taskBD" pool-size="1-10" queue-capacity="100" rejection-policy="CALLER_RUNS"/>

Hello. Программа читает все файлы в каталоге, разделяет страницы, преобразует информацию (основную или уникальную) и записывает информацию в базу.

У меня проблема с этим конфигом.

06-03-2019 13:01:24.773 [task-scheduler-7] ERROR o.s.i.handler.LoggingHandler.handleMessageInternal - org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@19c520db[Running, pool size = 5, active threads = 5, queued tasks = 200, completed tasks = 1353]] did not accept task: org.springframework.integration.util.ErrorHandlingTaskExecutor$$Lambda$241/1065676784@6991f347

Что я делаю не так? Если я не использую taskExecutor, программа работает. Но мне нужно увеличить скорость исполнения. В другой конфигурации работы <, но у меня есть эта проблема: <code>ERROR o.s.i.handler.LoggingHandler.handleMessageInternal - java.lang.OutOfMemoryError: Java heap space Спасибо за помощь.

Answer 1

Следует учитывать, что не следует использовать каналы очереди между ними. Достаточно просто распределить результаты опроса в taskExecutor из опроса.

Вы можете покинуть канал очереди после splitter, поскольку после разделения у вас будет один ко многим, поэтому было бы здорово сделать их параллельными. Хотя вы также можете использовать вместо него ExecutorChannel.

Непонятно, однако, почему у вас есть отклоненная задача, поскольку ваша политика CALLER_RUNS ...