Как сделать блок метода ThreadPoolExecutor submit (), если он насыщен?

Я хочу создать ThreadPoolExecutorтакой, чтобы при достижении максимального размера и заполнении очереди submit()метод блокировался при попытке добавить новые задачи. Нужно ли мне реализовать RejectedExecutionHandlerдля этого индивидуальный заказ или есть способ сделать это с помощью стандартной библиотеки Java?

Одно из возможных решений, которое я только что нашел:

public class BoundedExecutor {
    private final Executor exec;
    private final Semaphore semaphore;

    public BoundedExecutor(Executor exec, int bound) {
        this.exec = exec;
        this.semaphore = new Semaphore(bound);
    }

    public void submitTask(final Runnable command)
            throws InterruptedException, RejectedExecutionException {
        semaphore.acquire();
        try {
            exec.execute(new Runnable() {
                public void run() {
                    try {
                        command.run();
                    } finally {
                        semaphore.release();
                    }
                }
            });
        } catch (RejectedExecutionException e) {
            semaphore.release();
            throw e;
        }
    }
}

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

Вы можете использовать ThreadPoolExecutor и blockingQueue:

public class ImageManager {
    BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(blockQueueSize);
    RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy();
    private ExecutorService executorService =  new ThreadPoolExecutor(numOfThread, numOfThread, 
        0L, TimeUnit.MILLISECONDS, blockingQueue, rejectedExecutionHandler);

    private int downloadThumbnail(String fileListPath){
        executorService.submit(new yourRunnable());
    }
}

Вы должны использовать CallerRunsPolicy, который выполняет отклоненную задачу в вызывающем потоке. Таким образом, он не может отправлять какие-либо новые задачи исполнителю, пока эта задача не будет выполнена, после чего появятся несколько свободных потоков пула или процесс повторится.

http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/ThreadPoolExecutor.CallerRunsPolicy.html

Из документов:

Отклоненные задачи

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

1. В ThreadPoolExecutor.AbortPolicy по умолчанию обработчик выдает исключение среды выполнения RejectedExecutionException при отклонении.
2. В ThreadPoolExecutor.CallerRunsPolicy поток, который вызывает выполнение, сам запускает задачу. Это обеспечивает простой механизм управления обратной связью, который замедляет скорость отправки новых задач.
3. В ThreadPoolExecutor.DiscardPolicy задача, которая не может быть выполнена, просто удаляется.
4. В ThreadPoolExecutor.DiscardOldestPolicy, если исполнитель не выключен, задача во главе рабочей очереди удаляется, а затем выполняется повторная попытка выполнения (что может снова закончиться неудачей, что приведет к повторению этого).

Кроме того, при вызове ThreadPoolExecutorконструктора обязательно используйте ограниченную очередь, такую ​​как ArrayBlockingQueue . Иначе ничего не будет отклонено.

Изменить: в ответ на ваш комментарий установите размер ArrayBlockingQueue равным максимальному размеру пула потоков и используйте AbortPolicy.

Изменить 2: Хорошо, я понимаю, к чему вы клоните. Как насчет этого: переопределить beforeExecute()метод, чтобы проверить, что getActiveCount()не превышает getMaximumPoolSize(), и если это так, спать и повторить попытку?

Hibernate имеет BlockPolicyпростой элемент, который может делать то, что вы хотите:

См .: Executors.java

/**
 * A handler for rejected tasks that will have the caller block until
 * space is available.
 */
public static class BlockPolicy implements RejectedExecutionHandler {

    /**
     * Creates a <tt>BlockPolicy</tt>.
     */
    public BlockPolicy() { }

    /**
     * Puts the Runnable to the blocking queue, effectively blocking
     * the delegating thread until space is available.
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        try {
            e.getQueue().put( r );
        }
        catch (InterruptedException e1) {
            log.error( "Work discarded, thread was interrupted while waiting for space to schedule: {}", r );
        }
    }
}

Приведенный BoundedExecutorвыше ответ из Java Concurrency in Practice работает правильно только в том случае, если вы используете неограниченную очередь для Executor или если размер семафора не превышает размер очереди. Семафор является общим состоянием между отправляющим потоком и потоками в пуле, что делает возможным насыщение исполнителя, даже если размер очереди <bound <= (размер очереди + размер пула).

Использование CallerRunsPolicyдопустимо только в том случае, если ваши задачи не выполняются вечно, и в этом случае ваш поток отправки останется rejectedExecutionнавсегда, и плохая идея, если ваши задачи требуют много времени для выполнения, потому что поток отправки не может отправить какие-либо новые задачи или делать что-нибудь еще, если он сам выполняет задачу.

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

Я знаю, что это взлом, но, на мой взгляд, самый чистый из предложенных здесь ;-)

Поскольку ThreadPoolExecutor использует блокирующую очередь "предложение" вместо "положить", позволяет переопределить поведение "предложения" блокирующей очереди:

class BlockingQueueHack<T> extends ArrayBlockingQueue<T> {

    BlockingQueueHack(int size) {
        super(size);
    }

    public boolean offer(T task) {
        try {
            this.put(task);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return true;
    }
}

ThreadPoolExecutor tp = new ThreadPoolExecutor(1, 2, 1, TimeUnit.MINUTES, new BlockingQueueHack(5));

Я тестировал, и вроде работает. Реализация некоторой политики тайм-аута оставлена ​​как упражнение читателя.

Следующий класс оборачивается вокруг ThreadPoolExecutor и использует семафор для блокировки, когда рабочая очередь заполнена:

public final class BlockingExecutor { 

    private final Executor executor;
    private final Semaphore semaphore;

    public BlockingExecutor(int queueSize, int corePoolSize, int maxPoolSize, int keepAliveTime, TimeUnit unit, ThreadFactory factory) {
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
        this.executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, unit, queue, factory);
        this.semaphore = new Semaphore(queueSize + maxPoolSize);
    }

    private void execImpl (final Runnable command) throws InterruptedException {
        semaphore.acquire();
        try {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        command.run();
                    } finally {
                        semaphore.release();
                    }
                }
            });
        } catch (RejectedExecutionException e) {
            // will never be thrown with an unbounded buffer (LinkedBlockingQueue)
            semaphore.release();
            throw e;
        }
    }

    public void execute (Runnable command) throws InterruptedException {
        execImpl(command);
    }
}

Этот класс-оболочка основан на решении, приведенном в книге Брайана Гетца в книге Java Concurrency in Practice. Решение в книге принимает только два параметра конструктора: an Executorи границу, используемую для семафора. Это показано в ответе Fixpoint. У такого подхода есть проблема: он может попасть в состояние, когда потоки пула заняты, очередь заполнена, но семафор только что выпустил разрешение. ( semaphore.release()в блоке finally). В этом состоянии новая задача может получить только что выпущенное разрешение, но будет отклонена, поскольку очередь задач заполнена. Конечно, вы этого не хотите; вы хотите заблокировать в этом случае.

Чтобы решить эту проблему, мы должны использовать неограниченную очередь, как ясно упоминает JCiP. Семафор действует как защита, создавая эффект размера виртуальной очереди. Это имеет побочный эффект, заключающийся в том, что модуль может содержать maxPoolSize + virtualQueueSize + maxPoolSizeзадачи. Это почему? Из- semaphore.release()за блока в finally. Если все потоки пула вызывают этот оператор одновременно, то maxPoolSizeразрешения освобождаются, позволяя одному и тому же количеству задач войти в модуль. Если бы мы использовали ограниченную очередь, она все равно была бы заполнена, что привело бы к отклонению задачи. Теперь, поскольку мы знаем, что это происходит только тогда, когда поток пула почти завершен, это не проблема. Мы знаем, что поток пула не будет блокироваться, поэтому задача скоро будет снята из очереди.

Однако вы можете использовать ограниченную очередь. Только убедитесь, что его размер равен virtualQueueSize + maxPoolSize. Большие размеры бесполезны, семафор не позволит впустить больше элементов. Меньшие размеры приведут к отклонению задач. Вероятность отклонения задач увеличивается с уменьшением размера. Например, предположим, что вам нужен ограниченный исполнитель с maxPoolSize = 2 и virtualQueueSize = 5. Затем возьмите семафор с 5 + 2 = 7 разрешениями и фактическим размером очереди 5 + 2 = 7. Реальное количество задач, которые могут быть в отряде, тогда 2 + 5 + 2 = 9. Когда исполнитель заполнен (5 задач в очереди, 2 в пуле потоков, поэтому доступно 0 разрешений) и ВСЕ потоки пула освобождают свои разрешения, то входящие задачи могут получить ровно 2 разрешения.

Теперь решение от JCiP несколько громоздко в использовании, поскольку оно не обеспечивает соблюдение всех этих ограничений (неограниченная очередь или ограниченная этими математическими ограничениями и т. Д.). Я думаю, что это служит только хорошим примером, демонстрирующим, как вы можете создавать новые потокобезопасные классы на основе уже доступных частей, но не как полноценный, многоразовый класс. Не думаю, что последнее было намерением автора.

вы можете использовать собственный RejectedExecutionHandler, например

ThreadPoolExecutor tp= new ThreadPoolExecutor(core_size, // core size
                max_handlers, // max size 
                timeout_in_seconds, // idle timeout 
                TimeUnit.SECONDS, queue, new RejectedExecutionHandler() {
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        // This will block if the queue is full
                        try {
                            executor.getQueue().put(r);
                        } catch (InterruptedException e) {
                            System.err.println(e.getMessage());
                        }

                    }
                });

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

Я верю, что это даст вам желаемое поведение блокировки. Обработчик отказа никогда не будет соответствовать требованиям, поскольку это указывает на то, что исполнитель не может выполнить задачу. Что я мог представить, так это то, что вы получаете некоторую форму «занятого ожидания» в обработчике. Это не то, что вам нужно, вам нужна очередь для исполнителя, которая блокирует вызывающего ...

Чтобы избежать проблем с решением @FixPoint. Можно использовать ListeningExecutorService и освободить семафор onSuccess и onFailure внутри FutureCallback.

Недавно я обнаружил, что этот вопрос имеет ту же проблему. OP не говорит об этом явно, но мы не хотим использовать, RejectedExecutionHandlerкоторый выполняет задачу в потоке отправителя, потому что это будет недостаточно использовать рабочие потоки, если эта задача является длительной.

Прочитав все ответы и комментарии, в частности ошибочное решение с семафором или его использование, afterExecuteя внимательно изучил код ThreadPoolExecutor, чтобы увидеть, есть ли выход. Я был поражен, увидев, что существует более 2000 строк (прокомментированного) кода, некоторые из которых вызывают у меня головокружение . Учитывая довольно простое требование, которое у меня действительно есть - один производитель, несколько потребителей, пусть производитель блокирует, когда ни один из потребителей не может выполнять работу - я решил развернуть свое собственное решение. Это не файл, ExecutorServiceа просто файл Executor. И он не адаптирует количество потоков к рабочей нагрузке, а поддерживает только фиксированное количество потоков, что также соответствует моим требованиям. Вот код. Не стесняйтесь разглагольствовать об этом :-)

package x;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;

/**
 * distributes {@code Runnable}s to a fixed number of threads. To keep the
 * code lean, this is not an {@code ExecutorService}. In particular there is
 * only very simple support to shut this executor down.
 */
public class ParallelExecutor implements Executor {
  // other bounded queues work as well and are useful to buffer peak loads
  private final BlockingQueue<Runnable> workQueue =
      new SynchronousQueue<Runnable>();
  private final Thread[] threads;

  /*+**********************************************************************/
  /**
   * creates the requested number of threads and starts them to wait for
   * incoming work
   */
  public ParallelExecutor(int numThreads) {
    this.threads = new Thread[numThreads];
    for(int i=0; i<numThreads; i++) {
      // could reuse the same Runner all over, but keep it simple
      Thread t = new Thread(new Runner());
      this.threads[i] = t;
      t.start();
    }
  }
  /*+**********************************************************************/
  /**
   * returns immediately without waiting for the task to be finished, but may
   * block if all worker threads are busy.
   * 
   * @throws RejectedExecutionException if we got interrupted while waiting
   *         for a free worker
   */
  @Override
  public void execute(Runnable task)  {
    try {
      workQueue.put(task);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new RejectedExecutionException("interrupt while waiting for a free "
          + "worker.", e);
    }
  }
  /*+**********************************************************************/
  /**
   * Interrupts all workers and joins them. Tasks susceptible to an interrupt
   * will preempt their work. Blocks until the last thread surrendered.
   */
  public void interruptAndJoinAll() throws InterruptedException {
    for(Thread t : threads) {
      t.interrupt();
    }
    for(Thread t : threads) {
      t.join();
    }
  }
  /*+**********************************************************************/
  private final class Runner implements Runnable {
    @Override
    public void run() {
      while (!Thread.currentThread().isInterrupted()) {
        Runnable task;
        try {
          task = workQueue.take();
        } catch (InterruptedException e) {
          // canonical handling despite exiting right away
          Thread.currentThread().interrupt(); 
          return;
        }
        try {
          task.run();
        } catch (RuntimeException e) {
          // production code to use a logging framework
          e.printStackTrace();
        }
      }
    }
  }
}

Я считаю, что есть довольно элегантный способ решить эту проблему, используя java.util.concurrent.Semaphoreи делегируя поведение Executor.newFixedThreadPool. Новая служба исполнителя будет выполнять новую задачу, только если для этого есть поток. Блокирование управляется семафором с количеством разрешений, равным количеству потоков. Когда задача завершена, она возвращает разрешение.

public class FixedThreadBlockingExecutorService extends AbstractExecutorService {

private final ExecutorService executor;
private final Semaphore blockExecution;

public FixedThreadBlockingExecutorService(int nTreads) {
    this.executor = Executors.newFixedThreadPool(nTreads);
    blockExecution = new Semaphore(nTreads);
}

@Override
public void shutdown() {
    executor.shutdown();
}

@Override
public List<Runnable> shutdownNow() {
    return executor.shutdownNow();
}

@Override
public boolean isShutdown() {
    return executor.isShutdown();
}

@Override
public boolean isTerminated() {
    return executor.isTerminated();
}

@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
    return executor.awaitTermination(timeout, unit);
}

@Override
public void execute(Runnable command) {
    blockExecution.acquireUninterruptibly();
    executor.execute(() -> {
        try {
            command.run();
        } finally {
            blockExecution.release();
        }
    });
}

У меня была такая же потребность в прошлом: своего рода блокирующая очередь с фиксированным размером для каждого клиента, поддерживаемая общим пулом потоков. В итоге я написал свой собственный ThreadPoolExecutor:

UserThreadPoolExecutor (очередь блокировки (для каждого клиента) + пул потоков (общий для всех клиентов))

См .: https://github.com/d4rxh4wx/UserThreadPoolExecutor

Каждому UserThreadPoolExecutor предоставляется максимальное количество потоков из общего ThreadPoolExecutor.

Каждый UserThreadPoolExecutor может:

• отправить задачу исполнителю общего пула потоков, если его квота не достигнута. Если его квота достигнута, задание ставится в очередь (непотребляющая блокировка, ожидающая процессора). Как только одна из отправленных задач завершена, квота уменьшается, позволяя другой задаче, ожидающей передачи, быть отправленной в ThreadPoolExecutor.
• дождитесь завершения остальных задач

Я нашел эту политику отказа в клиенте эластичного поиска. Он блокирует вызывающий поток в очереди блокировки. Код ниже-

 static class ForceQueuePolicy implements XRejectedExecutionHandler 
 {
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) 
        {
            try 
            {
                executor.getQueue().put(r);
            } 
            catch (InterruptedException e) 
            {
                //should never happen since we never wait
                throw new EsRejectedExecutionException(e);
            }
        }

        @Override
        public long rejected() 
        {
            return 0;
        }
}

Недавно мне потребовалось добиться чего-то подобного, но на ScheduledExecutorService.

Я также должен был убедиться, что я обрабатываю задержку, передаваемую методу, и гарантирую, что либо задача будет отправлена ​​для выполнения в то время, которое ожидает вызывающий объект, либо просто не удастся, таким образом, бросив файл RejectedExecutionException.

Другие методы из ScheduledThreadPoolExecutorдля выполнения или отправки задачи изнутри вызывают, #scheduleкоторые, в свою очередь, будут вызывать переопределенные методы.

import java.util.concurrent.*;

public class BlockingScheduler extends ScheduledThreadPoolExecutor {
    private final Semaphore maxQueueSize;

    public BlockingScheduler(int corePoolSize,
                             ThreadFactory threadFactory,
                             int maxQueueSize) {
        super(corePoolSize, threadFactory, new AbortPolicy());
        this.maxQueueSize = new Semaphore(maxQueueSize);
    }

    @Override
    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
        final long newDelayInMs = beforeSchedule(command, unit.toMillis(delay));
        return super.schedule(command, newDelayInMs, TimeUnit.MILLISECONDS);
    }

    @Override
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay,
                                           TimeUnit unit) {
        final long newDelayInMs = beforeSchedule(callable, unit.toMillis(delay));
        return super.schedule(callable, newDelayInMs, TimeUnit.MILLISECONDS);
    }

    @Override
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {
        final long newDelayInMs = beforeSchedule(command, unit.toMillis(initialDelay));
        return super.scheduleAtFixedRate(command, newDelayInMs, unit.toMillis(period), TimeUnit.MILLISECONDS);
    }

    @Override
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long period,
                                                     TimeUnit unit) {
        final long newDelayInMs = beforeSchedule(command, unit.toMillis(initialDelay));
        return super.scheduleWithFixedDelay(command, newDelayInMs, unit.toMillis(period), TimeUnit.MILLISECONDS);
    }

    @Override
    protected void afterExecute(Runnable runnable, Throwable t) {
        super.afterExecute(runnable, t);
        try {
            if (t == null && runnable instanceof Future<?>) {
                try {
                    ((Future<?>) runnable).get();
                } catch (CancellationException | ExecutionException e) {
                    t = e;
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt(); // ignore/reset
                }
            }
            if (t != null) {
                System.err.println(t);
            }
        } finally {
            releaseQueueUsage();
        }
    }

    private long beforeSchedule(Runnable runnable, long delay) {
        try {
            return getQueuePermitAndModifiedDelay(delay);
        } catch (InterruptedException e) {
            getRejectedExecutionHandler().rejectedExecution(runnable, this);
            return 0;
        }
    }

    private long beforeSchedule(Callable callable, long delay) {
        try {
            return getQueuePermitAndModifiedDelay(delay);
        } catch (InterruptedException e) {
            getRejectedExecutionHandler().rejectedExecution(new FutureTask(callable), this);
            return 0;
        }
    }

    private long getQueuePermitAndModifiedDelay(long delay) throws InterruptedException {
        final long beforeAcquireTimeStamp = System.currentTimeMillis();
        maxQueueSize.tryAcquire(delay, TimeUnit.MILLISECONDS);
        final long afterAcquireTimeStamp = System.currentTimeMillis();
        return afterAcquireTimeStamp - beforeAcquireTimeStamp;
    }

    private void releaseQueueUsage() {
        maxQueueSize.release();
    }
}

У меня есть код, буду благодарен за любой отзыв. https://github.com/AmitabhAwasthi/BlockingScheduler

Вот решение, которое, кажется, действительно работает. Это называется NotifyingBlockingThreadPoolExecutor .

Демо-программа.

Изменить: есть проблема с этим кодом, метод await () ошибочен. Вызов shutdown () + awaitTermination (), похоже, работает нормально.

Мне не всегда нравится CallerRunsPolicy, тем более что он позволяет отклоненной задаче «пропускать очередь» и выполняться до задач, которые были отправлены ранее. Более того, выполнение задачи в вызывающем потоке может занять гораздо больше времени, чем ожидание освобождения первого слота.

Я решил эту проблему с помощью специального 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);
    }
}

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

executorPool = new ThreadPoolExecutor(1, 1, 10,
                                      TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
                                      new BlockWhenQueueFull());

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

Тем не менее лично мне этот метод нравится. Он компактен, прост для понимания и хорошо работает.