Блок ThreadPoolExecutor, когда очередь заполнена?

Question

Я пытаюсь выполнить множество задач с помощью ThreadPoolExecutor. Ниже приведен гипотетический пример:

def workQueue = new ArrayBlockingQueue<Runnable>(3, false) 
def threadPoolExecutor = new ThreadPoolExecutor(3, 3, 1L, TimeUnit.HOURS, workQueue) 
for(int i = 0; i < 100000; i++) 
    threadPoolExecutor.execute(runnable) 

Проблема заключается в том, что я быстро получить java.util.concurrent.RejectedExecutionException, так как число задач превышает размер очереди. Тем не менее, желаемое поведение, которое я ищу, состоит в том, чтобы иметь блок основного потока, пока в очереди не будет места. Каков наилучший способ сделать это?

Answer 1

В некоторых очень узких условиях вы можете реализовать java.util.concurrent.RejectedExecutionHandler, который делает то, что вам нужно.

RejectedExecutionHandler block = new RejectedExecutionHandler() { 
    rejectedExecution(Runnable r, ThreadPoolExecutor executor) { 
    executor.getQueue().put(r); 
    } 
}; 

ThreadPoolExecutor pool = new ... 
pool.setRejectedExecutionHandler(block); 

Теперь. Это очень плохая идея по следующим причинам

• Это склонное к тупиковому, потому что все потоки в бассейне могут умереть прежде, чем вещи вы кладете в очереди видно. Смягчите это, установив разумное время сохранения.
• Задача не завершена так, как может ожидать ваш Исполнитель. Множество реализаций исполнителей завершают свои задачи в каком-то объекте отслеживания перед выполнением. Посмотрите на свой источник.
• Добавление через getQueue() категорически отвергается API и в какой-то момент может быть запрещено.

Стратегия почти всегда лучше установить ThreadPoolExecutor.CallerRunsPolicy, которая будет дросселировать ваше приложение, запустив задачу в потоке, который вызывает execute().

Однако иногда стратегия блокировки со всеми вытекающими из нее рисками - это то, что вы хотите. Я бы сказал, что в этих условиях

• У вас есть только один поток вызова выполнить()
• Вы должны (или хотят) имеют очень малую длину очереди
• Вам абсолютно необходимо ограничить число потоки, выполняющие эту работу (как правило, по внешним причинам), и стратегия caller-run нарушит это.
• Ваши задачи имеют непредсказуемый размер, поэтому запуск звонков может вызвать голод, если пул на мгновение занят четырьмя короткими задачами, а ваш один поток, вызывающий выполнение, застрял с большим.

Итак, как я уже сказал. Это редко необходимо и может быть опасно, но там вы идете.

Удачи.

Answer 2

Вот мой фрагмент кода в этом случае:

public void executeBlocking(Runnable command) { 
    if (threadPool == null) { 
     logger.error("Thread pool '{}' not initialized.", threadPoolName); 
     return; 
    } 
    ThreadPool threadPoolMonitor = this; 
    boolean accepted = false; 
    do { 
     try { 
      threadPool.execute(new Runnable() { 
       @Override 
       public void run() { 
        try { 
         command.run(); 
        } 
        // to make sure that the monitor is freed on exit 
        finally { 
         // Notify all the threads waiting for the resource, if any. 
         synchronized (threadPoolMonitor) { 
          threadPoolMonitor.notifyAll(); 
         } 
        } 
       } 
      }); 
      accepted = true; 
     } 
     catch (RejectedExecutionException e) { 
      // Thread pool is full 
      try { 
       // Block until one of the threads finishes its job and exits. 
       synchronized (threadPoolMonitor) { 
        threadPoolMonitor.wait(); 
       } 
      } 
      catch (InterruptedException ignored) { 
       // return immediately 
       break; 
      } 
     } 
    } while (!accepted); 
} 

Threadpool является локальным экземпляром java.util.concurrent.ExecutorService, который был инициализирован уже.

Answer 3

Я решил эту проблему с помощью пользовательского RejectedExecutionHandler, который просто блокирует вызывающий поток на некоторое время, а затем снова пытается представить задачу:

public class BlockWhenQueueFull implements RejectedExecutionHandler { 

    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { 

     // The pool is full. Wait, then try again. 
     try { 
      long waitMs = 250; 
      Thread.sleep(waitMs); 
     } catch (InterruptedException interruptedException) {} 

     executor.execute(r); 
    } 
} 

Этот класс может только использоваться в резьбе бассейне исполнителем как RejectedExecutionHandler, как и любой другой. В этом примере:

executorPool = new def threadPoolExecutor = new ThreadPoolExecutor(3, 3, 1L, TimeUnit.HOURS, workQueue, new BlockWhenQueueFull()) 

Единственный недостаток я вижу в том, что вызывающий поток может получить немного заперта больше, чем строго необходимо (до 250 мс). Для многих коротких задач, возможно, уменьшите время ожидания до 10 мс или около того. Более того, поскольку этот исполнитель фактически называется рекурсивно, очень долго ждет, когда поток станет доступным (часы), может привести к переполнению стека.

Тем не менее, мне лично нравится этот метод. Он компактный, понятный и хорошо работает. Не хватает ли чего-нибудь важного?