Коротко — RejectedExecutionException возникает, когда при отправке задачи в ThreadPoolExecutor очередь заполнена и пул уже достиг $\text{maximumPoolSize}$ и использует политику отказа. Нужно проверить параметры пула, тип/ёмкость очереди, политику отказа и архитектуру подачи задач. Что проверить и что можно изменить:
Почему это происходит (одно предложение)
- Почти всегда: скорость поступления задач > скорость обработки и буфер (очередь + возможность создания потоков) исчерпан.
Проверить (метрики и состояния)
- Тип очереди и её ёмкость: $\text{workQueue}$ (LinkedBlockingQueue, ArrayBlockingQueue, SynchronousQueue) и её capacity / remainingCapacity.
- Параметры пула: $\text{corePoolSize}$, $\text{maximumPoolSize}$, $\text{keepAliveTime}$.
- Текущие показатели: $\text{poolSize}$, $\text{activeCount}$, $\text{queue.size()}$, $\text{completedTaskCount}$, число отказов (если логируете).
- Политика отказа: $\text{RejectedExecutionHandler}$ (Abort, CallerRuns, Discard, DiscardOldest или кастом).
- Поведение производителей: скорость отправки задач, всплески, retry/loop.
- Время выполнения задачи: среднее и распределение (\(\text{T_{cpu}}\), \(\text{T_{wait}}\)).
- Память/ресурсы: размер стека потока ($\text{stackSize}$) и лимит по памяти — большое число потоков может убить процесс.
- Проверить, не выключен ли Executor (isShutdown/isTerminated) и нет ли непойманных ошибок в задачах.
Как выбирать размеры пула (формула)
- Для CPU-bound: $\text{PoolSize}\approx N_{cores}\times \left(1+\frac{T_{wait}}{T_{cpu}}\right)$.
- Учитывайте память: приблизительно максимальное число потоков ограничено доступной памятью / размером стека ($\text{maxThreads}\approx \frac{\text{availableMemory}}{\text{stackSize}}$).
Конкретные изменения и архитектурные меры
- Использовать bounded очередь, не неограниченную LinkedBlockingQueue, чтобы получать backpressure: например ArrayBlockingQueue с разумной ёмкостью.
- Подобрать сочетание очереди и размеров пулa:
- Если нужна быстрая масштабируемость — SynchronousQueue + большой $\text{maximumPoolSize}$ (потоки создаются сразу, но нужен контроль по памяти).
- Если важна буферизация — bounded ArrayBlockingQueue и $\text{maximumPoolSize}>\text{corePoolSize}$.
- Политика отказа: чаще полезна $\text{CallerRunsPolicy}$ (технически замедляет производителя) или кастомная логика (лог + retry + метрика).
- Настроить $\text{keepAliveTime}$ и $\text{allowCoreThreadTimeOut(true)}$ для высвобождения лишних потоков после спада нагрузки.
- Ввести backpressure у производителей: семафор/ограничитель, который блокирует/тормозит отправку новых задач при заполнении буфера.
- Батчирование и декомпозиция задач: уменьшить число мелких задач или собирать их в пакеты.
- Перенос на асинхронные/неблокирующие библиотеки (Netty, reactive) для IO-bound нагрузок.
- Вынести в очередь сообщений (Kafka/RabbitMQ) для устойчивой обработки и разгрузки пиков.
- Горизонтальное масштабирование сервиса или количество потребителей.
- Использовать мониторинг и alert'ы по очередям и отказам (queueSize, activeCount, rejectedCount).
Практические рекомендации (коротко)
- Не используйте неограниченный LinkedBlockingQueue если хотите, чтобы $\text{maximumPoolSize}$ работал.
- Для долгих блокирующих задач увеличивайте число потоков аккуратно и применяйте bounded queue + CallerRunsPolicy или semaphore-based throttling.
- Для коротких CPU-bound задач используйте формулу выше и ForkJoinPool / work-stealing.
- Логируйте и метрикуйте все отклонения задач и размер очереди.
Пример быстрой конфигурации (смысл, не код)
- $\text{corePoolSize}=20$, $\text{maximumPoolSize}=200$, $\text{workQueue.capacity}=1000$, $\text{keepAliveTime}=60\text{s}$, $\text{RejectedExecutionHandler}=\text{CallerRunsPolicy}$ — и добавить мониторинг и throttling на отправителях.
Заключение
- Проверить тип и capacity очереди, pool sizes, политику отказа, профиль задач и скорость поступления. Ввести bounded буфер + backpressure или внешнюю очередь для пиков — это наиболее устойчивые архитектурные меры.