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

102

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?

Fixpunkt
quelle
2
Wollen Sie etwas wie die Angebotsmethode ()
Extraneon
Mögliches Duplikat von Java: ExecutorService, das bei
Übermittlung
2
@ Bacar Ich bin anderer Meinung. Diese Fragen und Antworten sehen wertvoller aus (zusätzlich dazu, dass sie älter sind).
JasonMArcher

Antworten:

47

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.

Fixpunkt
quelle
2
Gibt es hier eine Racebedingung zwischen dem Zeitpunkt, an dem das Semaphor in der finally-Klausel freigegeben wird und dem Erwerb des Semaphors?
Volni
2
Wie oben erwähnt, ist diese Implementierung fehlerhaft, da das Semaphor freigegeben wird, bevor die Aufgabe abgeschlossen ist. Es wäre besser, die Methode java.util.concurrent.ThreadPoolExecutor # afterExecute (Runnable, Throwable)
FelixM
2
@FelixM: Die Verwendung von java.util.concurrent.ThreadPoolExecutor # afterExecute (ausführbar, auslösbar) löst das Problem nicht, da afterExecute unmittelbar nach task.run () in java.util.concurrent.ThreadPoolExecutor # runWorker (Worker w) aufgerufen wird Nehmen Sie das nächste Element aus der Warteschlange (siehe Quellcode von openjdk 1.7.0.6).
Jaan
1
Diese Antwort stammt aus dem Buch Java Concurrency in Practice von Brian Goetz
orangepips
11
Diese Antwort ist nicht ganz richtig, ebenso die Kommentare. Dieser Code stammt in der Tat aus Java Concurrency in der Praxis, und es ist richtig, wenn Sie den Kontext berücksichtigen . In dem Buch heißt es wörtlich: "Verwenden Sie bei einem solchen Ansatz eine unbegrenzte Warteschlange (...) und setzen Sie die Grenze für das Semaphor so, dass sie der Poolgröße plus der Anzahl der Aufgaben in der Warteschlange entspricht, die Sie zulassen möchten." Bei einer unbegrenzten Warteschlange werden Aufgaben niemals abgelehnt, sodass das erneute Auslösen der Ausnahme völlig nutzlos ist! Welches ist, glaube ich, auch der Grund , warum throw e;ist nicht im Buch. JCIP ist richtig!
Timmos
30

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());
    }
}
DiTime4Tea
quelle
Ich möchte nur sagen, dass dies eine wahnsinnig schnelle und einfache Lösung war, die sehr gut funktioniert hat!
Ivan
58
Dadurch werden abgelehnte Aufgaben im übergebenden Thread ausgeführt. Was funktional nicht den Anforderungen des OP entspricht.
Wahrnehmung
4
Dadurch wird die Aufgabe "im aufrufenden Thread" ausgeführt, anstatt sie zu blockieren, um sie in die Warteschlange zu stellen. Dies kann einige nachteilige Auswirkungen haben, z. B. wenn mehrere Threads sie auf diese Weise aufrufen, werden mehr als "Jobs mit Warteschlangengröße" ausgeführt, und wenn Die Aufgabe dauert länger als erwartet. Ihr "produzierender" Thread hält den Executor möglicherweise nicht beschäftigt. Aber hier hat super funktioniert!
Rogerdpack
4
Downvoted: Dies wird nicht blockiert, wenn das TPE gesättigt ist. Dies ist nur eine Alternative, keine Lösung.
Timmos
1
Upvoted: Dies passt ziemlich gut zum 'Design von TPE' und 'blockiert natürlich' Client-Threads, indem es ihnen Überlauf-Tasts gibt. Dies sollte die meisten Anwendungsfälle abdecken, aber natürlich nicht alle, und Sie sollten verstehen, was unter der Haube vor sich geht.
Mike
12

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.

danben
quelle
3
Ich möchte, dass die Anzahl der gleichzeitig ausgeführten Aufgaben streng begrenzt wird (durch die Anzahl der Threads in Executor). Aus diesem Grund kann ich nicht zulassen, dass Aufrufer-Threads diese Aufgaben selbst ausführen.
Fixpunkt
1
AbortPolicy würde dazu führen, dass der Aufruferthread eine RejectedExecutionException erhält, während ich sie nur zum Blockieren benötige.
Fixpunkt
2
Ich bitte um Blockierung, nicht Schlaf & Polling;)
Fixpoint
@danben: Meinst du nicht CallerRunsPolicy ?
user359996
7
Das Problem mit CallerRunPolicy besteht darin, dass bei einem einzelnen Thread-Produzenten häufig Threads nicht verwendet werden, wenn eine Aufgabe mit langer Laufzeit abgelehnt wird (da die anderen Aufgaben im Thread-Pool beendet werden, während die Aufgabe mit langer Laufzeit noch ausgeführt wird Laufen).
Adam Gent
6

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 );
        }
    }
}
Nate Murray
quelle
4
Auf den zweiten Blick ist dies eine ziemlich schlechte Idee. Ich empfehle Ihnen nicht, es zu verwenden. Für gute Gründe siehe hier: stackoverflow.com/questions/3446011/…
Nate Murray
Dies verwendet auch nicht die "Standard-Java-Bibliothek" gemäß der Anforderung des OP. Löschen?
user359996
1
Woah, das ist so hässlich. Grundsätzlich stört diese Lösung die Interna von TPE. Das Javadoc für ThreadPoolExecutorsagt sogar wörtlich: "Die Methode getQueue () ermöglicht den Zugriff auf die Arbeitswarteschlange zum Zwecke der Überwachung und des Debuggens. Von der Verwendung dieser Methode für andere Zwecke wird dringend abgeraten." Dass dies in einer so weithin bekannten Bibliothek verfügbar ist, ist absolut traurig zu sehen.
Timmos
1
com.amazonaws.services.simpleworkflow.flow.worker.BlockCallerPolicy ist ähnlich.
Adrian Baker
6

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.

stephan f
quelle
Ich bin nicht sicher, wie das Überprüfen der Warteschlangenlänge vor dem Senden keine abgelehnten Aufgaben in einer Multithread-Umgebung mit mehreren Aufgabenherstellern garantiert. Das klingt nicht threadsicher.
Tim
5

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.

Jakub
quelle
Eine bereinigte Version davon finden Sie unter stackoverflow.com/a/4522411/2601671 . Ich stimme zu, es ist der sauberste Weg, es zu tun.
Trenton
3

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.

Timmos
quelle
2

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

                    }
                });
Amir David Nissan Cohen
quelle
1
In den Dokumenten für getQueue () wird ausdrücklich erwähnt, dass der Zugriff auf
Chadi
0

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 ...

Fried Hoeben
quelle
2
ThreadPoolExecutorVerwendet eine offerMethode, um der Warteschlange Aufgaben hinzuzufügen. Wenn ich einen benutzerdefinierten Benutzer BlockingQueueerstellen offerwürde , der blockiert , würde dies den BlockingQueueVertrag brechen .
Fixpunkt
@Shooshpanchick, das würde den BlockingQueues-Vertrag brechen. Na und? Wenn Sie so interessiert sind, können Sie das Verhalten explizit nur während submit ()
aktivieren
Siehe auch diese Antwort auf eine andere Frage, die diese Alternative beschreibt.
Robert Tupelo-Schneck
Gibt es einen Grund, warum ThreadPoolExecutordie Verwendung implementiert wurde offerund nicht put(die blockierende Version)? Wenn es eine Möglichkeit für den Client-Code gäbe, zu sagen, welche wann verwendet werden soll, wären viele Leute, die versuchen, benutzerdefinierte Lösungen von Hand zu rollen, erleichtert gewesen
wie
0

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

vamsu
quelle
Dies hat die gleichen inhärenten Probleme wie das Umschließen der RunnableMethoden, da diese Methoden vor der normalen Bereinigung der Worker noch aufgerufen werden ThreadPoolExecutor. Das heißt, Sie müssen weiterhin Ablehnungsausnahmen behandeln.
Adam Gent
0

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

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();
        }
    });
}
Radoslaw Kachel
quelle
Ich habe den in Java Concurrency in der Praxis beschriebenen BoundedExecutor implementiert und festgestellt, dass das Semaphor mit dem Fairness-Flag true initialisiert werden muss, um sicherzustellen, dass Semaphor-Genehmigungen in den Auftragsanforderungen angeboten werden. Siehe docs.oracle.com/javase/7/docs/api/java/util/concurrent/… . für Details
Prahalad Deshpande
0

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
d4rxh4wx
quelle
0

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

Dev Amitabh
quelle
Diese Antwort basiert vollständig auf dem Inhalt externer Links. Sollten sie jemals ungültig werden, wäre Ihre Antwort nutzlos. Bearbeiten Sie also bitte Ihre Antwort und fügen Sie mindestens eine Zusammenfassung dessen hinzu, was dort zu finden ist. Danke dir!
Fabio sagt Reinstate Monica
@fabio: danke für den Hinweis. Ich habe den Code dort hinzugefügt, damit er für die Leser jetzt sinnvoller ist. Schätzen Sie Ihren Kommentar :)
Dev Amitabh
0

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.

Stefan Reich
quelle
0

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.

TinkerTank
quelle
1
Wie Sie selbst sagen: Dies kann einen Stapelüberlauf erzeugen. Nicht etwas, das ich im Produktionscode haben möchte.
Harald
Jeder sollte seine eigenen Entscheidungen treffen. Für meine Arbeitsbelastung ist dies kein Problem. Aufgaben werden in Sekunden anstatt in den Stunden ausgeführt, die zum Aufblasen des Stapels erforderlich wären. Darüber hinaus kann das Gleiche für praktisch jeden rekursiven Algorithmus gesagt werden. Ist das ein Grund, niemals einen rekursiven Algorithmus in der Produktion zu verwenden?
TinkerTank