Wie wird der Submit () -Methodenblock von ThreadPoolExecutor blockiert, wenn er gesättigt ist?

Ich möchte eine ThreadPoolExecutorsolche erstellen , dass die submit()Methode blockiert, wenn versucht wird, neue Aufgaben hinzuzufügen , wenn die maximale Größe erreicht ist und die Warteschlange voll ist . Muss ich dafür eine benutzerdefinierte Implementierung implementieren RejectedExecutionHandleroder gibt es eine Möglichkeit, dies mithilfe einer Standard-Java-Bibliothek zu tun?

Eine der möglichen Lösungen, die ich gerade gefunden habe:

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;
        }
    }
}

Gibt es noch andere Lösungen? Ich würde etwas bevorzugen, das darauf basiert, RejectedExecutionHandlerda es eine Standardmethode zu sein scheint, um mit solchen Situationen umzugehen.

Sie können ThreadPoolExecutor und eine blockierende Warteschlange verwenden:

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());
    }
}

Sie sollten das verwenden CallerRunsPolicy, das die abgelehnte Aufgabe im aufrufenden Thread ausführt. Auf diese Weise können keine neuen Aufgaben an den Executor gesendet werden, bis diese Aufgabe erledigt ist. Zu diesem Zeitpunkt gibt es einige freie Pool-Threads oder der Vorgang wird wiederholt.

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

Aus den Dokumenten:

Abgelehnte Aufgaben

Neue Aufgaben, die in der Methode execute (java.lang.Runnable) übergeben wurden, werden abgelehnt, wenn der Executor heruntergefahren wurde und wenn der Executor endliche Grenzen für maximale Threads und Arbeitswarteschlangenkapazität verwendet und gesättigt ist. In beiden Fällen ruft die Methode execute die Methode RejectedExecutionHandler.rejectedExecution (java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) ihres RejectedExecutionHandler auf. Es stehen vier vordefinierte Handlerrichtlinien zur Verfügung:

1. Im Standard-ThreadPoolExecutor.AbortPolicy löst der Handler bei Ablehnung eine Laufzeit-RejectedExecutionException aus.
2. In ThreadPoolExecutor.CallerRunsPolicy führt der aufgerufene Thread die Aufgabe selbst aus. Dies bietet einen einfachen Mechanismus zur Steuerung der Rückmeldung, der die Übermittlungsrate neuer Aufgaben verlangsamt.
3. In ThreadPoolExecutor.DiscardPolicy wird eine Aufgabe, die nicht ausgeführt werden kann, einfach gelöscht.
4. Wenn in ThreadPoolExecutor.DiscardOldestPolicy der Executor nicht heruntergefahren wird, wird die Aufgabe am Kopf der Arbeitswarteschlange gelöscht und die Ausführung erneut versucht (was erneut fehlschlagen kann und dazu führt, dass dies wiederholt wird).

Stellen Sie außerdem sicher, dass Sie beim Aufrufen des ThreadPoolExecutorKonstruktors eine begrenzte Warteschlange wie ArrayBlockingQueue verwenden. Andernfalls wird nichts abgelehnt.

Bearbeiten: Stellen Sie als Antwort auf Ihren Kommentar die Größe der ArrayBlockingQueue so ein, dass sie der maximalen Größe des Thread-Pools entspricht, und verwenden Sie AbortPolicy.

Edit 2: Ok, ich verstehe, worauf du hinaus willst. Was ist damit: Überschreiben Sie die beforeExecute()Methode, um zu überprüfen, ob sie getActiveCount()nicht überschritten getMaximumPoolSize()wird. Wenn dies der Fall ist, schlafen Sie und versuchen Sie es erneut.

Hibernate hat eine BlockPolicy, die einfach ist und tun kann, was Sie wollen:

Siehe: 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 );
        }
    }
}

Die BoundedExecutoroben aus Java Concurrency in Practice zitierte Antwort funktioniert nur dann ordnungsgemäß, wenn Sie eine unbegrenzte Warteschlange für den Executor verwenden oder die gebundene Semaphorgröße nicht größer als die Warteschlangengröße ist. Das Semaphor wird vom übergebenden Thread und den Threads im Pool gemeinsam genutzt, sodass der Executor auch dann gesättigt werden kann, wenn die Warteschlangengröße <gebunden <= (Warteschlangengröße + Poolgröße) ist.

Die Verwendung CallerRunsPolicyist nur gültig, wenn Ihre Aufgaben nicht für immer ausgeführt werden. In diesem Fall bleibt Ihr übermittelnder Thread für rejectedExecutionimmer erhalten, und eine schlechte Idee, wenn die Ausführung Ihrer Aufgaben lange dauert, da der übermittelnde Thread keine neuen Aufgaben übermitteln kann oder Tun Sie etwas anderes, wenn eine Aufgabe selbst ausgeführt wird.

Wenn dies nicht akzeptabel ist, empfehle ich, die Größe der begrenzten Warteschlange des Executors zu überprüfen, bevor Sie eine Aufgabe senden. Wenn die Warteschlange voll ist, warten Sie eine kurze Zeit, bevor Sie erneut versuchen, sie zu senden. Der Durchsatz wird darunter leiden, aber ich schlage vor, es ist eine einfachere Lösung als viele der anderen vorgeschlagenen Lösungen, und Sie werden garantiert, dass keine Aufgaben abgelehnt werden.

Ich weiß, es ist ein Hack, aber meiner Meinung nach der sauberste Hack zwischen den hier angebotenen ;-)

Da ThreadPoolExecutor die Blockierungswarteschlange "Angebot" anstelle von "Put" verwendet, können Sie das Verhalten von "Angebot" der Blockierungswarteschlange überschreiben:

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));

Ich habe es getestet und es scheint zu funktionieren. Das Implementieren einer Timeout-Richtlinie bleibt als Übung für den Leser übrig.

Die folgende Klasse umschließt einen ThreadPoolExecutor und blockiert mithilfe eines Semaphors, wenn die Arbeitswarteschlange voll ist:

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);
    }
}

Diese Wrapper-Klasse basiert auf einer Lösung aus dem Buch Java Concurrency in Practice von Brian Goetz. Die Lösung in diesem Buch verwendet nur zwei Konstruktorparameter: eine Executorund eine für das Semaphor verwendete Grenze. Dies wird in der Antwort von Fixpoint gezeigt. Bei diesem Ansatz gibt es ein Problem: Er kann sich in einem Zustand befinden, in dem die Pool-Threads ausgelastet sind, die Warteschlange voll ist, das Semaphor jedoch gerade eine Genehmigung freigegeben hat. (semaphore.release() im letzten Block). In diesem Status kann eine neue Aufgabe die gerade freigegebene Berechtigung abrufen, wird jedoch abgelehnt, da die Aufgabenwarteschlange voll ist. Natürlich ist das nicht etwas, was du willst; Sie möchten in diesem Fall blockieren.

Um dies zu lösen, müssen wir eine unbegrenzte Warteschlange verwenden, wie JCiP klar erwähnt. Das Semaphor fungiert als Schutz und wirkt wie eine virtuelle Warteschlange. Dies hat den Nebeneffekt, dass das Gerät möglicherweise maxPoolSize + virtualQueueSize + maxPoolSizeAufgaben enthalten kann. Warum ist das so? Wegen der semaphore.release()im Endblock. Wenn alle Pool-Threads diese Anweisung gleichzeitig aufrufen, werden maxPoolSizeGenehmigungen freigegeben, sodass dieselbe Anzahl von Aufgaben in die Einheit gelangen kann. Wenn wir eine begrenzte Warteschlange verwenden würden, wäre diese immer noch voll, was zu einer abgelehnten Aufgabe führen würde. Da wir wissen, dass dies nur auftritt, wenn ein Pool-Thread fast fertig ist, ist dies kein Problem. Wir wissen, dass der Pool-Thread nicht blockiert wird, sodass bald eine Aufgabe aus der Warteschlange entfernt wird.

Sie können jedoch eine begrenzte Warteschlange verwenden. Stellen Sie einfach sicher, dass die Größe gleich ist virtualQueueSize + maxPoolSize. Größere Größen sind nutzlos. Das Semaphor verhindert, dass mehr Elemente eingelassen werden. Kleinere Größen führen zu abgelehnten Aufgaben. Die Wahrscheinlichkeit, dass Aufgaben abgelehnt werden, steigt mit abnehmender Größe. Angenommen, Sie möchten einen begrenzten Executor mit maxPoolSize = 2 und virtualQueueSize = 5. Nehmen Sie dann ein Semaphor mit 5 + 2 = 7 Genehmigungen und einer tatsächlichen Warteschlangengröße von 5 + 2 = 7. Die tatsächliche Anzahl der Aufgaben, die in der Einheit ausgeführt werden können, beträgt dann 2 + 5 + 2 = 9. Wenn der Executor voll ist (5 Aufgaben in der Warteschlange, 2 im Thread-Pool, also 0 Berechtigungen verfügbar) und ALLE Pool-Threads ihre Berechtigungen freigeben, können genau 2 Genehmigungen von eingehenden Aufgaben übernommen werden.

Jetzt ist die Lösung von JCiP etwas umständlich zu verwenden, da sie nicht alle diese Einschränkungen erzwingt (unbegrenzte Warteschlange oder mit diesen mathematischen Einschränkungen verbunden usw.). Ich denke, dass dies nur ein gutes Beispiel ist, um zu demonstrieren, wie Sie neue thread-sichere Klassen basierend auf den bereits verfügbaren Teilen erstellen können, aber nicht als ausgewachsene, wiederverwendbare Klasse. Ich glaube nicht, dass Letzteres die Absicht des Autors war.

Sie können einen benutzerdefinierten RejectedExecutionHandler wie diesen verwenden

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());
                        }

                    }
                });

Erstellen Sie Ihre eigene Blockierungswarteschlange, die vom Executor mit dem gewünschten Blockierungsverhalten verwendet werden soll, und geben Sie dabei immer die verfügbare verbleibende Kapazität zurück (stellen Sie sicher, dass der Executor nicht versucht, mehr Threads als seinen Kernpool zu erstellen oder den Ablehnungshandler auszulösen).

Ich glaube, dies bringt Ihnen das Blockierungsverhalten, das Sie suchen. Ein Ablehnungshandler passt niemals in die Rechnung, da dies anzeigt, dass der Executor die Aufgabe nicht ausführen kann. Was ich mir dort vorstellen kann, ist, dass Sie im Handler eine Art "beschäftigtes Warten" haben. Das ist nicht was Sie wollen, Sie wollen eine Warteschlange für den Executor, die den Anrufer blockiert ...

Um Probleme mit der @ FixPoint-Lösung zu vermeiden. Man könnte ListeningExecutorService verwenden und das Semaphor onSuccess und onFailure in FutureCallback freigeben.

Kürzlich fand ich diese Frage mit dem gleichen Problem. Das OP sagt dies nicht explizit, aber wir möchten nicht das verwenden, RejectedExecutionHandlerdas eine Aufgabe im Thread des Übermittlers ausführt, da dadurch die Arbeitsthreads nicht ausreichend genutzt werden, wenn diese Aufgabe lange ausgeführt wird.

Beim Lesen aller Antworten und Kommentare, insbesondere der fehlerhaften Lösung mit dem Semaphor oder beim Verwenden, habe afterExecuteich mir den Code des ThreadPoolExecutor genauer angesehen, um festzustellen , ob es einen Ausweg gibt. Ich war erstaunt zu sehen, dass es mehr als 2000 Zeilen (kommentierten) Codes gibt, von denen mir einige schwindelig werden . Angesichts der ziemlich einfachen Anforderung, die ich tatsächlich habe - ein Hersteller, mehrere Verbraucher, lassen Sie den Hersteller blockieren, wenn kein Verbraucher arbeiten kann -, habe ich beschlossen, meine eigene Lösung zu entwickeln. Es ist kein, ExecutorServicesondern nur ein Executor. Und es passt die Anzahl der Threads nicht an die Arbeitslast an, sondern enthält nur eine feste Anzahl von Threads, was auch meinen Anforderungen entspricht. Hier ist der Code. Fühlen Sie sich frei, darüber zu schimpfen :-)

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();
        }
      }
    }
  }
}

Ich glaube, es gibt eine ziemlich elegante Möglichkeit, dieses Problem zu lösen, indem das java.util.concurrent.SemaphoreVerhalten von verwendet und delegiert wird Executor.newFixedThreadPool. Der neue Executor-Dienst führt eine neue Aufgabe nur aus, wenn ein Thread dafür vorhanden ist. Das Blockieren wird von Semaphore mit einer Anzahl von Genehmigungen verwaltet, die der Anzahl von Threads entspricht. Wenn eine Aufgabe abgeschlossen ist, wird eine Genehmigung zurückgegeben.

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();
        }
    });
}

Ich hatte in der Vergangenheit das gleiche Bedürfnis: eine Art Blockierungswarteschlange mit einer festen Größe für jeden Client, der von einem gemeinsam genutzten Thread-Pool unterstützt wird. Am Ende habe ich meine eigene Art von ThreadPoolExecutor geschrieben:

UserThreadPoolExecutor (Blockierungswarteschlange (pro Client) + Threadpool (von allen Clients gemeinsam genutzt))

Siehe: https://github.com/d4rxh4wx/UserThreadPoolExecutor

Jeder UserThreadPoolExecutor erhält eine maximale Anzahl von Threads von einem gemeinsam genutzten ThreadPoolExecutor

Jeder UserThreadPoolExecutor kann:

• Senden Sie eine Aufgabe an den Executor des gemeinsam genutzten Thread-Pools, wenn dessen Kontingent nicht erreicht wird. Wenn sein Kontingent erreicht ist, wird der Job in die Warteschlange gestellt (nicht verbrauchendes Blockieren wartet auf CPU). Sobald eine der übergebenen Aufgaben abgeschlossen ist, wird das Kontingent dekrementiert, sodass eine weitere Aufgabe, die darauf wartet, an den ThreadPoolExecutor gesendet zu werden
• Warten Sie, bis die verbleibenden Aufgaben abgeschlossen sind

Ich habe diese Ablehnungsrichtlinie im elastischen Suchclient gefunden. Es blockiert den Anrufer-Thread in der Blockierungswarteschlange. Code unten-

 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;
        }
}

Ich hatte vor kurzem das Bedürfnis, etwas Ähnliches zu erreichen, aber auf einem ScheduledExecutorService.

Ich musste auch sicherstellen, dass ich die Verzögerung handele, die an die Methode übergeben wird, und sicherstellen, dass entweder die Aufgabe zur Ausführung gesendet wird, wie es der Aufrufer erwartet, oder einfach fehlschlägt, wodurch a ausgelöst wird RejectedExecutionException.

Andere Methoden ScheduledThreadPoolExecutorzum Ausführen oder Senden einer Aufgabe rufen intern auf, #schedulewodurch wiederum die überschriebenen Methoden aufgerufen werden.

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();
    }
}

Ich habe den Code hier, freue mich über jedes Feedback. https://github.com/AmitabhAwasthi/BlockingScheduler

Hier ist die Lösung, die wirklich gut zu funktionieren scheint. Es heißt NotifyingBlockingThreadPoolExecutor .

Demo-Programm.

Bearbeiten: Es gibt ein Problem mit diesem Code, die await () -Methode ist fehlerhaft. Das Aufrufen von shutdown () + awaitTermination () scheint gut zu funktionieren.

Ich mag CallerRunsPolicy nicht immer, zumal es der abgelehnten Aufgabe ermöglicht, die Warteschlange zu überspringen und vor Aufgaben ausgeführt zu werden, die zuvor gesendet wurden. Darüber hinaus kann das Ausführen der Aufgabe im aufrufenden Thread viel länger dauern, als darauf zu warten, dass der erste Steckplatz verfügbar wird.

Ich habe dieses Problem mit einem benutzerdefinierten RejectedExecutionHandler gelöst, der den aufrufenden Thread einfach für eine Weile blockiert und dann versucht, die Aufgabe erneut zu senden:

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);
    }
}

Diese Klasse kann im Thread-Pool-Executor einfach wie jede andere als RejectedExecutinHandler verwendet werden, zum Beispiel:

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

Der einzige Nachteil, den ich sehe, ist, dass der aufrufende Thread möglicherweise etwas länger als unbedingt erforderlich gesperrt wird (bis zu 250 ms). Da dieser Executor effektiv rekursiv aufgerufen wird, kann eine sehr lange Wartezeit auf die Verfügbarkeit eines Threads (Stunden) zu einem Stapelüberlauf führen.

Trotzdem gefällt mir diese Methode persönlich. Es ist kompakt, leicht zu verstehen und funktioniert gut.