Es ist unmöglich, einen zwischengespeicherten Thread-Pool mit einer Größenbeschränkung zu erstellen?

127

Es scheint unmöglich zu sein, einen zwischengespeicherten Thread-Pool mit einer Begrenzung der Anzahl der Threads zu erstellen, die er erstellen kann.

So wird statisches Executors.newCachedThreadPool in der Standard-Java-Bibliothek implementiert:

 public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

Verwenden Sie diese Vorlage, um einen zwischengespeicherten Thread-Pool mit fester Größe zu erstellen:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new SynchronusQueue<Runable>());

Wenn Sie dies jetzt verwenden und 3 Aufgaben einreichen, ist alles in Ordnung. Das Senden weiterer Aufgaben führt zu abgelehnten Ausführungsausnahmen.

Versuchen Sie dies:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runable>());

Wird dazu führen, dass alle Threads nacheinander ausgeführt werden. Das heißt, der Thread-Pool wird niemals mehr als einen Thread erstellen, um Ihre Aufgaben zu erledigen.

Dies ist ein Fehler in der Ausführungsmethode von ThreadPoolExecutor? Oder ist das vielleicht beabsichtigt? Oder gibt es einen anderen Weg?

Bearbeiten: Ich möchte genau so etwas wie den zwischengespeicherten Thread-Pool (er erstellt Threads bei Bedarf und beendet sie dann nach einer gewissen Zeit), jedoch mit einer Begrenzung der Anzahl der Threads, die er erstellen kann, und der Möglichkeit, weitere Aufgaben in die Warteschlange zu stellen, sobald er vorhanden ist das Thread-Limit erreicht. Nach der Antwort von sjlee ist dies unmöglich. Wenn man sich die execute () -Methode von ThreadPoolExecutor ansieht, ist dies in der Tat unmöglich. Ich müsste ThreadPoolExecutor in eine Unterklasse unterteilen und execute () überschreiben, ähnlich wie SwingWorker, aber was SwingWorker in seiner execute () tut, ist ein vollständiger Hack.

Matt Crinklaw-Vogt
quelle
1
Was ist deine Frage? Ist Ihr zweites Codebeispiel nicht die Antwort auf Ihren Titel?
rsp
4
Ich möchte einen Thread-Pool, der bei Bedarf Threads hinzufügt, wenn die Anzahl der Aufgaben zunimmt, aber niemals mehr als eine maximale Anzahl von Threads hinzufügt. CachedThreadPool führt dies bereits aus, außer dass eine unbegrenzte Anzahl von Threads hinzugefügt wird und nicht bei einer vordefinierten Größe angehalten wird. Die Größe, die ich in den Beispielen definiere, ist 3. Im zweiten Beispiel wird 1 Thread hinzugefügt, aber nicht zwei weitere, wenn neue Aufgaben eintreffen, während die anderen Aufgaben noch nicht abgeschlossen sind.
Matt Crinklaw-Vogt
Überprüfen Sie dies, es löst es, debuggingisfun.blogspot.com/2012/05/…
Ethan

Antworten:

235

Der ThreadPoolExecutor weist die folgenden Hauptverhaltensweisen auf, und Ihre Probleme können durch diese Verhaltensweisen erklärt werden.

Wenn Aufgaben eingereicht werden,

  1. Wenn der Thread-Pool die Kerngröße nicht erreicht hat, werden neue Threads erstellt.
  2. Wenn die Kerngröße erreicht wurde und keine inaktiven Threads vorhanden sind, werden Aufgaben in die Warteschlange gestellt.
  3. Wenn die Kerngröße erreicht wurde, gibt es keine inaktiven Threads und die Warteschlange wird voll. Es werden neue Threads erstellt (bis die maximale Größe erreicht ist).
  4. Wenn die maximale Größe erreicht wurde, gibt es keine inaktiven Threads und die Warteschlange wird voll. Die Ablehnungsrichtlinie wird aktiviert.

Beachten Sie im ersten Beispiel, dass die SynchronousQueue im Wesentlichen die Größe 0 hat. Sobald Sie die maximale Größe (3) erreicht haben, wird die Ablehnungsrichtlinie aktiviert (# 4).

Im zweiten Beispiel ist die Warteschlange der Wahl eine LinkedBlockingQueue mit einer unbegrenzten Größe. Daher bleiben Sie bei Verhalten Nr. 2 hängen.

Sie können nicht wirklich viel mit dem zwischengespeicherten Typ oder dem festen Typ basteln, da deren Verhalten fast vollständig bestimmt ist.

Wenn Sie einen begrenzten und dynamischen Thread-Pool haben möchten, müssen Sie eine positive Kerngröße und eine maximale Größe in Kombination mit einer Warteschlange endlicher Größe verwenden. Beispielsweise,

new ThreadPoolExecutor(10, // core size
    50, // max size
    10*60, // idle timeout
    TimeUnit.SECONDS,
    new ArrayBlockingQueue<Runnable>(20)); // queue with a size

Nachtrag : Dies ist eine ziemlich alte Antwort, und es scheint, dass JDK sein Verhalten in Bezug auf die Kerngröße 0 geändert hat. Seit JDK 1.6 fügt der ThreadPoolExecutor a hinzu, wenn die Kerngröße 0 ist und der Pool keine Threads enthält Thread, um diese Aufgabe auszuführen. Daher ist die Kerngröße 0 eine Ausnahme von der obigen Regel. Dank Steve für bringen , dass meine Aufmerksamkeit.

sjlee
quelle
4
Sie müssen einige Wörter über die Methode schreiben allowCoreThreadTimeOut, um diese Antwort perfekt zu machen. Siehe die Antwort von @ user1046052
hsestupin
1
Gute Antwort! Nur ein Punkt zum Hinzufügen: Andere Ablehnungsrichtlinien sind ebenfalls erwähnenswert. Siehe die Antwort von @brianegge
Jeff
1
Sollte Verhalten 2 nicht sagen: "Wenn die maxThread- Größe erreicht wurde und keine inaktiven Threads vorhanden sind, werden Aufgaben in die Warteschlange gestellt." ?
Zoltán
1
Könnten Sie näher erläutern, was die Größe der Warteschlange impliziert? Bedeutet dies, dass nur 20 Aufgaben in die Warteschlange gestellt werden können, bevor sie abgelehnt werden?
Zoltán
1
@ Zoltán Ich habe dies vor einiger Zeit geschrieben, daher besteht die Möglichkeit, dass sich seitdem ein gewisses Verhalten geändert hat (ich habe die neuesten Aktivitäten nicht zu genau verfolgt), aber unter der Annahme, dass dieses Verhalten unverändert ist, ist # 2 wie angegeben korrekt, und das ist es vielleicht der wichtigste (und etwas überraschende) Punkt davon. Sobald die Kerngröße erreicht ist, bevorzugt TPE das Anstehen gegenüber dem Erstellen neuer Threads. Die Warteschlangengröße entspricht buchstäblich der Größe der Warteschlange, die an das TPE übergeben wird. Wenn die Warteschlange voll wird, aber die maximale Größe nicht erreicht hat, wird ein neuer Thread erstellt (keine Aufgaben ablehnen). Siehe # 3. Hoffentlich hilft das.
sjlee
60

Sofern ich nichts verpasst habe, ist die Lösung der ursprünglichen Frage einfach. Der folgende Code implementiert das gewünschte Verhalten, wie im Originalposter beschrieben. Es werden bis zu 5 Threads erzeugt, um an einer unbegrenzten Warteschlange zu arbeiten, und inaktive Threads werden nach 60 Sekunden beendet.

tp = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<Runnable>());
tp.allowCoreThreadTimeOut(true);

quelle
1
Du hast Recht. Diese Methode wurde in jdk 1.6 hinzugefügt, sodass nicht so viele Leute davon wissen. Außerdem können Sie keine "minimale" Kernpoolgröße haben, was unglücklich ist.
Jtahlborn
4
Meine einzige Sorge ist (aus den JDK 8-Dokumenten): "Wenn eine neue Aufgabe in der Methode execute (Runnable) übergeben wird und weniger als corePoolSize-Threads ausgeführt werden, wird ein neuer Thread erstellt, um die Anforderung zu verarbeiten, selbst wenn andere Mitarbeiter Threads sind inaktiv. "
Veegee
Ich bin mir ziemlich sicher, dass das nicht funktioniert. Das letzte Mal, als ich mir das oben Gesagte angesehen habe, wird Ihre Arbeit tatsächlich immer nur in einem Thread ausgeführt, obwohl Sie 5 erzeugt haben. Auch hier waren es einige Jahre, aber als ich mich mit der Implementierung von ThreadPoolExecutor befasste, wurde sie erst dann an neue Threads gesendet, wenn Ihre Warteschlange voll war. Die Verwendung einer unbegrenzten Warteschlange führt dazu, dass dies niemals geschieht. Sie können testen, indem Sie Arbeit einreichen und den Thread-Namen protokollieren und dann schlafen. Jede ausführbare Datei druckt am Ende denselben Namen / wird nicht in einem anderen Thread ausgeführt.
Matt Crinklaw-Vogt
2
Das funktioniert, Matt. Sie haben die Kerngröße auf 0 gesetzt, deshalb hatten Sie nur 1 Thread. Der Trick dabei ist, die Kerngröße auf die maximale Größe einzustellen.
T-Gergely
1
@vegee ist richtig - Dies funktioniert nicht wirklich gut - ThreadPoolExecutor verwendet Threads nur wieder, wenn sie über corePoolSize liegen. Wenn corePoolSize also gleich maxPoolSize ist, profitieren Sie nur dann vom Thread-Caching, wenn Ihr Pool voll ist. Wenn Sie dies also verwenden möchten, aber normalerweise unter Ihrer maximalen Poolgröße bleiben, können Sie das Thread-Timeout genauso gut auf einen niedrigen Wert reduzieren Wert; und seien Sie sich bewusst, dass es kein Caching gibt - immer neue Threads)
Chris Riddell
7

Hatte das gleiche Problem. Da keine andere Antwort alle Fragen zusammenfasst, füge ich meine hinzu:

Es ist jetzt klar in Dokumenten geschrieben : Wenn Sie eine Warteschlange verwenden, die die LinkedBlockingQueueEinstellung für ( ) max-Threads nicht blockiert , hat dies keine Auswirkung, werden nur Kernthreads verwendet.

so:

public class MyExecutor extends ThreadPoolExecutor {

    public MyExecutor() {
        super(4, 4, 5,TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
        allowCoreThreadTimeOut(true);
    }

    public void setThreads(int n){
        setMaximumPoolSize(Math.max(1, n));
        setCorePoolSize(Math.max(1, n));
    }

}

Dieser Testamentsvollstrecker hat:

  1. Kein Konzept für maximale Threads, da wir eine unbegrenzte Warteschlange verwenden. Dies ist eine gute Sache, da eine solche Warteschlange dazu führen kann, dass der Executor eine große Anzahl von nicht zum Kern gehörenden zusätzlichen Threads erstellt, wenn er seiner üblichen Richtlinie folgt.

  2. Eine Warteschlange mit maximaler Größe Integer.MAX_VALUE. Submit()wird geworfen, RejectedExecutionExceptionwenn die Anzahl der ausstehenden Aufgaben überschritten wird Integer.MAX_VALUE. Ich bin mir nicht sicher, ob uns zuerst der Speicher ausgeht, sonst passiert dies.

  3. Hat 4 Kerngewinde möglich. Leerlauf-Core-Threads werden automatisch beendet, wenn sie 5 Sekunden lang inaktiv sind. Also, ja, ausschließlich bei Bedarf. Threads können mithilfe der setThreads()Methode variiert werden .

  4. Stellt sicher, dass die minimale Anzahl von Kernthreads niemals kleiner als eins ist, da sonst submit()jede Aufgabe abgelehnt wird. Da Kernthreads> = max threads sein müssen, setThreads()legt die Methode auch max threads fest, obwohl die maximale Threadeinstellung für eine unbegrenzte Warteschlange unbrauchbar ist.

SD
quelle
Ich denke, Sie müssen auch 'allowCoreThreadTimeOut' auf 'true' setzen, andernfalls behalten Sie die Threads nach dem Erstellen
eric
Ups, ich habe das nur verpasst, sorry, deine Antwort ist dann perfekt!
Eric
6

In Ihrem ersten Beispiel werden nachfolgende Aufgaben abgelehnt, da dies AbortPolicydie Standardeinstellung ist RejectedExecutionHandler. Der ThreadPoolExecutor enthält die folgenden Richtlinien, die Sie über die setRejectedExecutionHandlerMethode ändern können :

CallerRunsPolicy
AbortPolicy
DiscardPolicy
DiscardOldestPolicy

Es hört sich so an, als ob Sie einen zwischengespeicherten Thread-Pool mit einer CallerRunsPolicy möchten.

Brianegge
quelle
5

Keine der Antworten hier hat mein Problem behoben, das mit dem Erstellen einer begrenzten Anzahl von HTTP-Verbindungen mit dem HTTP-Client von Apache (Version 3.x) zu tun hatte. Da ich einige Stunden gebraucht habe, um ein gutes Setup zu finden, werde ich Folgendes mitteilen:

private ExecutorService executor = new ThreadPoolExecutor(5, 10, 60L,
  TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
  Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());

Dadurch wird ein erstellt, ThreadPoolExecutorder mit fünf beginnt und maximal zehn gleichzeitig ausgeführte Threads enthält, die CallerRunsPolicyzur Ausführung verwendet werden.

paul
quelle
Das Problem bei dieser Lösung besteht darin, dass Sie die Anzahl der Threads erhöhen, auf denen die Hintergrund-Threads ausgeführt werden, wenn Sie die Anzahl oder die Produzenten erhöhen. In vielen Fällen ist das nicht das, was Sie wollen.
Grau
3

Per Javadoc für ThreadPoolExecutor:

Wenn mehr als corePoolSize-Threads, aber weniger als maximalPoolSize-Threads ausgeführt werden, wird nur dann ein neuer Thread erstellt , wenn die Warteschlange voll ist . Indem Sie corePoolSize und maximumPoolSize gleich einstellen, erstellen Sie einen Thread-Pool mit fester Größe.

(Hervorhebung von mir.)

Die Antwort von Jitter ist, was Sie wollen, obwohl meine Ihre andere Frage beantwortet. :) :)

Jonathan Feinberg
quelle
2

Es gibt noch eine Option. Anstatt die neue SynchronousQueue zu verwenden, können Sie auch jede andere Warteschlange verwenden. Sie müssen jedoch sicherstellen, dass die Größe 1 ist, damit der Executorservice gezwungen wird, einen neuen Thread zu erstellen.

Ashkrit
quelle
Ich denke, Sie meinen Größe 0 (standardmäßig), damit keine Aufgabe in die Warteschlange gestellt wird und der Executorservice jedes Mal gezwungen wird, einen neuen Thread zu erstellen.
Leonmax
2

Sieht nicht so aus, als ob eine der Antworten die Frage tatsächlich beantwortet - tatsächlich sehe ich keinen Weg, dies zu tun -, selbst wenn Sie eine Unterklasse von PooledExecutorService haben, da viele der Methoden / Eigenschaften privat sind, z Mach Folgendes:

class MyThreadPoolService extends ThreadPoolService {
    public void execute(Runnable run) {
        if (poolSize() == 0) {
            if (addIfUnderMaximumPoolSize(run) != null)
                return;
        }
        super.execute(run);
    }
}

Das nächste, was ich bekam, war das - aber selbst das ist keine sehr gute Lösung

new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()) {
    public void execute(Runnable command) {
        if (getPoolSize() == 0 && getActiveCount() < getMaximumPoolSize()) {        
            super.setCorePoolSize(super.getCorePoolSize() + 1);
        }
        super.execute(command);
    }

    protected void afterExecute(Runnable r, Throwable t) {
         // nothing in the queue
         if (getQueue().isEmpty() && getPoolSize() > min) {
             setCorePoolSize(getCorePoolSize() - 1);
         }
    };
 };

ps hat das oben nicht getestet

Stuart
quelle
2

Hier ist eine andere Lösung. Ich denke, diese Lösung verhält sich so, wie Sie es möchten (obwohl Sie nicht stolz auf diese Lösung sind):

final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {
    public boolean offer(Runnable o) {
        if (size() > 1)
            return false;
        return super.offer(o);
    };

    public boolean add(Runnable o) {
        if (super.offer(o))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }
};

RejectedExecutionHandler handler = new RejectedExecutionHandler() {         
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        queue.add(r);
    }
};

dbThreadExecutor =
        new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS, queue, handler);
Stuart
quelle
2

Das ist was du willst (zumindest denke ich das). Eine Erklärung finden Sie in der Antwort von Jonathan Feinberg

Executors.newFixedThreadPool(int n)

Erstellt einen Thread-Pool, der eine feste Anzahl von Threads wiederverwendet, die in einer gemeinsam genutzten unbegrenzten Warteschlange ausgeführt werden. Zu jedem Zeitpunkt sind höchstens nThreads-Threads aktive Verarbeitungsaufgaben. Wenn zusätzliche Aufgaben gesendet werden, während alle Threads aktiv sind, warten sie in der Warteschlange, bis ein Thread verfügbar ist. Wenn ein Thread aufgrund eines Fehlers während der Ausführung vor dem Herunterfahren beendet wird, wird bei Bedarf ein neuer Thread ersetzt, um nachfolgende Aufgaben auszuführen. Die Threads im Pool bleiben bestehen, bis sie explizit heruntergefahren werden.

Jitter
quelle
4
Sicher, ich könnte einen festen Thread-Pool verwenden, aber das würde n Threads für immer oder bis zum Aufruf von shutdown belassen. Ich möchte etwas genau wie den zwischengespeicherten Thread-Pool (er erstellt Threads bei Bedarf und beendet sie dann nach einer gewissen Zeitüberschreitung), jedoch mit einer Begrenzung der Anzahl der Threads, die er erstellen kann.
Matt Crinklaw-Vogt
0
  1. Sie können ThreadPoolExecutorwie von @sjlee vorgeschlagen verwenden

    Sie können die Größe des Pools dynamisch steuern. Schauen Sie sich diese Frage für weitere Details an:

    Dynamischer Thread-Pool

    ODER

  2. Sie können die newWorkStealingPool- API verwenden, die mit Java 8 eingeführt wurde.

    public static ExecutorService newWorkStealingPool()

    Erstellt einen Thread, der die Arbeit stiehlt, wobei alle verfügbaren Prozessoren als Zielparallelitätsstufe verwendet werden.

Standardmäßig ist die Parallelitätsstufe auf die Anzahl der CPU-Kerne in Ihrem Server eingestellt. Wenn Sie einen 4-Kern-CPU-Server haben, beträgt die Größe des Thread-Pools 4. Diese API gibt den ForkJoinPoolTyp von ExecutorService und ermöglicht das Stehlen von inaktiven Threads durch das Stehlen von Aufgaben von ausgelasteten Threads in ForkJoinPool.

Ravindra Babu
quelle
0

Das Problem wurde wie folgt zusammengefasst:

Ich möchte genau das Gleiche wie den zwischengespeicherten Thread-Pool (er erstellt Threads bei Bedarf und beendet sie dann nach einer gewissen Zeitüberschreitung), jedoch mit einer Begrenzung der Anzahl der Threads, die er erstellen kann, und der Möglichkeit, weitere Aufgaben in die Warteschlange zu stellen, sobald er erreicht ist Gewindebegrenzung.

Bevor ich auf die Lösung verweise, werde ich erklären, warum die folgenden Lösungen nicht funktionieren:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());

Dadurch werden keine Aufgaben in die Warteschlange gestellt, wenn das Limit von 3 erreicht ist, da SynchronousQueue per Definition keine Elemente enthalten kann.

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

Dadurch wird nicht mehr als ein einzelner Thread erstellt, da ThreadPoolExecutor nur dann Threads erstellt, die die corePoolSize überschreiten, wenn die Warteschlange voll ist. LinkedBlockingQueue ist jedoch niemals voll.

ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 3, 60, TimeUnit.SECONDS,
    new LinkedBlockingQueue<Runnable>());
executor.allowCoreThreadTimeOut(true);

Dadurch werden Threads erst wiederverwendet, wenn corePoolSize erreicht wurde, da ThreadPoolExecutor die Anzahl der Threads erhöht, bis corePoolSize erreicht ist, auch wenn vorhandene Threads inaktiv sind. Wenn Sie mit diesem Nachteil leben können, ist dies die einfachste Lösung für das Problem. Dies ist auch die in "Java Concurrency in Practice" (Fußnote auf S. 172) beschriebene Lösung.

Die einzige vollständige Lösung für das beschriebene Problem scheint darin zu bestehen, die offerMethode der Warteschlange zu überschreiben und eine zu schreiben, RejectedExecutionHandlerwie in den Antworten auf diese Frage erläutert: Wie kann der ThreadPoolExecutor die Anzahl der Threads vor dem Anstehen auf maximal erhöhen?

Stefan Feuerhahn
quelle
0

Dies funktioniert für Java8 + (und andere, vorerst ..)

     Executor executor = new ThreadPoolExecutor(3, 3, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>()){{allowCoreThreadTimeOut(true);}};

Dabei ist 3 die Grenze für die Anzahl der Threads und 5 das Zeitlimit für inaktive Threads.

Wenn Sie überprüfen möchten, ob es selbst funktioniert, finden Sie hier den Code für die Ausführung der Aufgabe:

public static void main(String[] args) throws InterruptedException {
    final int DESIRED_NUMBER_OF_THREADS=3; // limit of number of Threads for the task at a time
    final int DESIRED_THREAD_IDLE_DEATH_TIMEOUT=5; //any idle Thread ends if it remains idle for X seconds

    System.out.println( java.lang.Thread.activeCount() + " threads");
    Executor executor = new ThreadPoolExecutor(DESIRED_NUMBER_OF_THREADS, DESIRED_NUMBER_OF_THREADS, DESIRED_THREAD_IDLE_DEATH_TIMEOUT, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>()) {{allowCoreThreadTimeOut(true);}};

    System.out.println(java.lang.Thread.activeCount() + " threads");

    for (int i = 0; i < 5; i++) {
        final int fi = i;
        executor.execute(() -> waitsout("starting hard thread computation " + fi, "hard thread computation done " + fi,2000));
    }
    System.out.println("If this is UP, it works");

    while (true) {
        System.out.println(
                java.lang.Thread.activeCount() + " threads");
        Thread.sleep(700);
    }

}

static void waitsout(String pre, String post, int timeout) {
    try {
        System.out.println(pre);
        Thread.sleep(timeout);
        System.out.println(post);
    } catch (Exception e) {
    }
}

Ausgabe des obigen Codes ist für mich

1 threads
1 threads
If this is UP, it works
starting hard thread computation 0
4 threads
starting hard thread computation 2
starting hard thread computation 1
4 threads
4 threads
hard thread computation done 2
hard thread computation done 0
hard thread computation done 1
starting hard thread computation 3
starting hard thread computation 4
4 threads
4 threads
4 threads
hard thread computation done 3
hard thread computation done 4
4 threads
4 threads
4 threads
4 threads
3 threads
3 threads
3 threads
1 threads
1 threads
Ondřej
quelle