Что может быть причиной исключения RejectedExecutionException

Я получаю это исключение на моем сервере tomcat (+ liferay)

java.util.concurrent.RejectedExecutionException

мой класс выглядит следующим образом:

public class SingleExecutor extends ThreadPoolExecutor {
  public SingleExecutor(){
    super(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
  }

  @Override
  public void execute(Runnable command) {
    if(command instanceof AccessLogInsert){
        AccessLogInsert ali = (AccessLogInsert)command;
        ali.setConn(conn);
        ali.setPs(ps);
    }
    super.execute(command);
  }
}

Я получаю это исключение в строке super.execute(command); Эта ошибка может возникнуть, когда очередь заполнена, но размер LinkedBlockingQueue равен 2 ^ 31, и я уверен, что ожидания команды не так много.

В начале все стабильно, но после того, как я переделаю войну, она начинается. Этот класс не является частью войны, а в банке в tomcat/lib.

У вас есть идея, почему это произошло и как это исправить?

Ответ 1

От ThreadPoolExecutor JavaDoc

Новые задачи, отправленные в методе execute(java.lang.Runnable), будут отклонены при отключении Executor, а также когда Executor использует конечные границы как для максимальных потоков, так и для рабочей очереди и является насыщенным. В любом случае метод execute вызывает метод RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) его RejectedExecutionHandler. Предоставляются четыре предопределенные политики обработчика:

В стандартном ThreadPoolExecutor.AbortPolicy обработчик выдает время выполнения RejectedExecutionException при отказе.

В ThreadPoolExecutor.CallerRunsPolicy поток, который вызывает сам выполнение, запускает задачу. Это обеспечивает простой механизм управления обратной связью, который замедляет скорость отправки новых задач.

В ThreadPoolExecutor.DiscardPolicy задача, которая не может быть выполнена, просто удаляется.

В ThreadPoolExecutor.DiscardOldestPolicy, если исполнитель не закрыт, задача во главе рабочей очереди отбрасывается, а затем выполняется повторное выполнение (что может снова потерпеть неудачу, в результате чего это будет повторяться.)
Можно определить и использовать другие классы RejectedExecutionHandler. Для этого требуется определенная осторожность, особенно когда политики предназначены для работы только в определенных режимах или политике очередей.

Предположительно поэтому перезагрузка войны вызывает остановку Executor. Попробуйте разместить соответствующие библиотеки на войне, чтобы Tomcat ClassLoader имел больше шансов правильно перезагрузить приложение.

Ответ 2

Просто чтобы добавить в OrangeDog отличный ответ, контракт Executor действительно таков, что его метод execute будет генерировать RejectedExecutionException когда исполнитель насыщен (т.е. в очереди нет места).

Однако было бы полезно, если бы он блокировался вместо этого, автоматически ожидая, пока в очереди не останется места для новой задачи.

С помощью следующего пользовательского BlockingQueue можно добиться этого:

public final class ThreadPoolQueue extends ArrayBlockingQueue<Runnable> {

    public ThreadPoolQueue(int capacity) {
        super(capacity);
    }

    @Override
    public boolean offer(Runnable e) {
        try {
            put(e);
        } catch (InterruptedException e1) {
            return false;
        }
        return true;
    }

}

Это, по существу, реализует алгоритм противодавления, замедляя производителя всякий раз, когда исполнитель насыщается.

Используйте это как:

int n = Runtime.getRuntime().availableProcessors();
ThreadPoolExecutor executor = new ThreadPoolExecutor(0, n, 1, TimeUnit.MINUTES, new ThreadPoolQueue(n));
for (Runnable task : tasks) {
    executor.execute(task); // will never throw, nor will queue more than n tasks
}
executor.shutdown();
executor.awaitTermination(1, TimeUnit.HOURS);