Wie bringt man den ThreadPoolExecutor dazu, die Anzahl der Threads vor dem Einreihen auf das Maximum zu erhöhen?

99

Ich bin seit einiger Zeit frustriert über das Standardverhalten, ThreadPoolExecutordas die ExecutorServiceThread-Pools unterstützt, die so viele von uns verwenden. Um aus den Javadocs zu zitieren:

Wenn mehr als corePoolSize, aber weniger als maximalPoolSize-Threads ausgeführt werden, wird ein neuer Thread nur erstellt , wenn die Warteschlange voll ist .

Dies bedeutet, dass beim Definieren eines Thread-Pools mit dem folgenden Code der zweite Thread niemals gestartet wird, da der Thread LinkedBlockingQueueunbegrenzt ist.

ExecutorService threadPool =
   new ThreadPoolExecutor(1 /*core*/, 50 /*max*/, 60 /*timeout*/,
      TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(/* unlimited queue */));

Nur wenn Sie eine begrenzte Warteschlange haben und die Warteschlange voll ist, werden Threads über der Kernnummer gestartet. Ich vermute, dass eine große Anzahl von Junior-Java-Multithread-Programmierern dieses Verhalten der nicht kennt ThreadPoolExecutor.

Jetzt habe ich einen speziellen Anwendungsfall, bei dem dies nicht optimal ist. Ich suche nach Möglichkeiten, ohne meine eigene TPE-Klasse zu schreiben, um das Problem zu umgehen.

Meine Anforderungen gelten für einen Webdienst, der Rückrufe an einen möglicherweise unzuverlässigen Dritten durchführt.

  • Ich möchte den Rückruf nicht synchron mit der Webanforderung durchführen, daher möchte ich einen Thread-Pool verwenden.
  • Normalerweise bekomme ich ein paar davon pro Minute, daher möchte ich keine newFixedThreadPool(...)mit einer großen Anzahl von Threads haben, die meistens inaktiv sind.
  • Von Zeit zu Zeit bekomme ich einen Ausbruch dieses Datenverkehrs und möchte die Anzahl der Threads auf einen Maximalwert erhöhen (sagen wir 50).
  • Ich muss den besten Versuch unternehmen, alle Rückrufe durchzuführen, damit ich weitere Rückrufe über 50 in die Warteschlange stellen kann. Ich möchte den Rest meines Webservers nicht mit a überfordern newCachedThreadPool().

Wie kann ich diese Einschränkung umgehen, ThreadPoolExecutorwenn die Warteschlange begrenzt und voll sein muss, bevor weitere Threads gestartet werden? Wie kann ich dafür sorgen, dass mehr Threads gestartet werden, bevor Aufgaben in die Warteschlange gestellt werden?

Bearbeiten:

@Flavio macht einen guten Punkt bei der Verwendung von ThreadPoolExecutor.allowCoreThreadTimeOut(true), um das Timeout und das Beenden der Kernthreads zu erreichen . Ich habe darüber nachgedacht, aber ich wollte immer noch die Core-Threads-Funktion. Ich wollte nicht, dass die Anzahl der Threads im Pool nach Möglichkeit unter die Kerngröße fällt.

Grau
quelle
1
Gibt es angesichts der Tatsache, dass in Ihrem Beispiel maximal 10 Threads erstellt werden, echte Einsparungen bei der Verwendung von etwas, das über einen Thread-Pool mit fester Größe wächst / schrumpft?
Bstempi
Guter Punkt @bstempi. Die Anzahl war etwas willkürlich. Ich habe es in der Frage auf 50 erhöht. Ich bin mir nicht sicher, wie viele gleichzeitige Threads ich tatsächlich arbeiten möchte, jetzt, wo ich diese Lösung habe.
Grau
1
Oh verdammt! 10 Gegenstimmen, wenn ich hier könnte, genau die gleiche Position, in der ich mich befinde.
Eugene

Antworten:

50

Wie kann ich diese Einschränkung umgehen, ThreadPoolExecutorwenn die Warteschlange begrenzt und voll sein muss, bevor weitere Threads gestartet werden?

Ich glaube, ich habe endlich eine etwas elegante (vielleicht etwas hackige) Lösung für diese Einschränkung gefunden ThreadPoolExecutor. Es geht um erstreckt , LinkedBlockingQueuees zu haben Rückkehr falsezu , queue.offer(...)wenn es bereits einige Tasks in der Liste. Wenn die aktuellen Threads nicht mit den Aufgaben in der Warteschlange Schritt halten, fügt das TPE zusätzliche Threads hinzu. Befindet sich der Pool bereits bei maximaler Anzahl von Threads, RejectedExecutionHandlerwird der aufgerufen. Es ist der Handler, der das put(...)in die Warteschlange stellt.

Es ist sicherlich seltsam, eine Warteschlange zu schreiben, offer(...)die zurückkehren kann falseund put()niemals blockiert, also ist das der Hack-Teil. Dies funktioniert jedoch gut mit der Verwendung der Warteschlange durch TPE, sodass ich kein Problem damit sehe.

Hier ist der Code:

// extend LinkedBlockingQueue to force offer() to return false conditionally
BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {
    private static final long serialVersionUID = -6903933921423432194L;
    @Override
    public boolean offer(Runnable e) {
        // Offer it to the queue if there is 0 items already queued, else
        // return false so the TPE will add another thread. If we return false
        // and max threads have been reached then the RejectedExecutionHandler
        // will be called which will do the put into the queue.
        if (size() == 0) {
            return super.offer(e);
        } else {
            return false;
        }
    }
};
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1 /*core*/, 50 /*max*/,
        60 /*secs*/, TimeUnit.SECONDS, queue);
threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        try {
            // This does the actual put into the queue. Once the max threads
            //  have been reached, the tasks will then queue up.
            executor.getQueue().put(r);
            // we do this after the put() to stop race conditions
            if (executor.isShutdown()) {
                throw new RejectedExecutionException(
                    "Task " + r + " rejected from " + e);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return;
        }
    }
});

Wenn ich mit diesem Mechanismus Aufgaben an die Warteschlange sende, wird der ThreadPoolExecutorWille:

  1. Skalieren Sie die Anzahl der Threads zunächst auf die Kerngröße (hier 1).
  2. Bieten Sie es der Warteschlange an. Wenn die Warteschlange leer ist, wird sie in die Warteschlange gestellt, damit sie von den vorhandenen Threads verarbeitet werden kann.
  3. Wenn die Warteschlange bereits 1 oder mehrere Elemente enthält, offer(...)wird false zurückgegeben.
  4. Wenn false zurückgegeben wird, skalieren Sie die Anzahl der Threads im Pool, bis sie die maximale Anzahl erreichen (hier 50).
  5. Wenn am Maximum, dann ruft es die RejectedExecutionHandler
  6. Die RejectedExecutionHandlerdann legt die Aufgabe in die Warteschlange durch den ersten verfügbaren Thread in FIFO - Reihenfolge verarbeitet werden.

Obwohl in meinem obigen Beispielcode die Warteschlange unbegrenzt ist, können Sie sie auch als begrenzte Warteschlange definieren. Wenn Sie beispielsweise eine Kapazität von 1000 hinzufügen LinkedBlockingQueue, wird Folgendes ausgeführt:

  1. Skalieren Sie die Gewinde auf max
  2. Stellen Sie sich dann in die Warteschlange, bis 1000 Aufgaben erledigt sind
  3. Blockieren Sie dann den Anrufer, bis der Warteschlange Speicherplatz zur Verfügung steht.

Auch, wenn Sie verwenden benötigt offer(...)in der RejectedExecutionHandlerdann könnte man die Verwendung offer(E, long, TimeUnit)Methode stattdessen mit Long.MAX_VALUEals Timeout.

Warnung:

Wenn Sie erwarten, dass dem Executor nach dem Herunterfahren Aufgaben hinzugefügt werden , sollten Sie klüger sein RejectedExecutionException, RejectedExecutionHandlerwenn Sie den Executor-Service heruntergefahren haben. Vielen Dank an @RaduToader für diesen Hinweis.

Bearbeiten:

Eine weitere Optimierung dieser Antwort könnte darin bestehen, das TPE zu fragen, ob Leerlauf-Threads vorhanden sind, und das Element nur dann in die Warteschlange zu stellen, wenn dies der Fall ist. Sie müssten dafür eine echte Klasse erstellen und eine ourQueue.setThreadPoolExecutor(tpe);Methode hinzufügen .

Dann offer(...)könnte Ihre Methode ungefähr so ​​aussehen:

  1. Überprüfen Sie, ob die tpe.getPoolSize() == tpe.getMaximumPoolSize()in diesem Fall nur anrufen super.offer(...).
  2. Andernfalls tpe.getPoolSize() > tpe.getActiveCount()rufen Sie an, super.offer(...)da es scheinbar inaktive Threads gibt.
  3. Andernfalls kehren Sie falsezu einem anderen Thread zurück.

Vielleicht das:

int poolSize = tpe.getPoolSize();
int maximumPoolSize = tpe.getMaximumPoolSize();
if (poolSize >= maximumPoolSize || poolSize > tpe.getActiveCount()) {
    return super.offer(e);
} else {
    return false;
}

Beachten Sie, dass die get-Methoden für TPE teuer sind, da sie auf volatileFelder zugreifen oder (im Fall von getActiveCount()) das TPE sperren und die Thread-Liste durchlaufen. Außerdem gibt es hier Rennbedingungen, die dazu führen können, dass eine Aufgabe nicht ordnungsgemäß in die Warteschlange gestellt oder ein anderer Thread gegabelt wird, wenn ein Leerlauf-Thread vorhanden ist.

Grau
quelle
Ich hatte auch mit dem gleichen Problem zu kämpfen und endete damit, die Ausführungsmethode zu überschreiben. Aber das ist wirklich eine schöne Lösung. :)
Batty
So sehr mir die Idee, den Vertrag zu brechen, Queueum dies zu erreichen, nicht gefällt , sind Sie mit Ihrer Idee sicherlich nicht allein: groovy-programming.com/post/26923146865
bstempi
3
Bekommst du hier nicht die Seltsamkeit, dass die ersten paar Aufgaben in die Warteschlange gestellt werden und erst danach neue Threads erscheinen? Wenn Ihr One-Core-Thread beispielsweise mit einer einzelnen Aufgabe mit langer Laufzeit beschäftigt ist und Sie anrufen execute(runnable), runnablewird er nur zur Warteschlange hinzugefügt. Wenn Sie anrufen execute(secondRunnable), secondRunnablewird dies zur Warteschlange hinzugefügt. Aber jetzt, wenn Sie anrufen execute(thirdRunnable), thirdRunnablewird in einem neuen Thread ausgeführt. Die runnableund secondRunnablenur einmalige Ausführung thirdRunnable(oder die ursprüngliche Langzeitaufgabe) sind beendet.
Robert Tupelo-Schneck
1
Ja, Robert hat Recht, in einer Umgebung mit starkem Multithreading wächst die Warteschlange manchmal, während freie Threads verwendet werden können. Die Lösung, unter der TPE erweitert wird, funktioniert viel besser. Ich denke, Roberts Vorschlag sollte als Antwort markiert werden, obwohl der obige Hack interessant ist
Wanna Know All
1
Der "RejectedExecutionHandler" half dem Executor beim Herunterfahren. Jetzt werden Sie gezwungen, shutdownNow () zu verwenden, da shutdown () das Hinzufügen neuer Aufgaben (aufgrund von Anforderungen) nicht verhindert
Radu Toader
28

Stellen Sie die Kerngröße und die maximale Größe auf denselben Wert ein und ermöglichen Sie das Entfernen von Kernthreads aus dem Pool mit allowCoreThreadTimeOut(true).

Flavio
quelle
+1 Ja, daran habe ich gedacht, aber ich wollte immer noch die Core-Threads-Funktion haben. Ich wollte nicht, dass der Thread-Pool in Ruhephasen auf 0 Threads gesetzt wird. Ich werde meine Frage bearbeiten, um darauf hinzuweisen. Aber ausgezeichneter Punkt.
Grau
Danke dir! Es ist einfach der einfachste Weg, das zu tun.
Dmitry Ovchinnikov
28

Ich habe bereits zwei weitere Antworten auf diese Frage, aber ich vermute, dass diese die beste ist.

Es basiert auf der Technik der aktuell akzeptierten Antwort , nämlich:

  1. Überschreiben Sie die offer()Methode der Warteschlange, um (manchmal) false zurückzugeben.
  2. was dazu führt, dass der ThreadPoolExecutorentweder einen neuen Thread erzeugt oder die Aufgabe ablehnt, und
  3. setzen die RejectedExecutionHandlerauf tatsächlich auf Ablehnung der Aufgabe Warteschlange.

Das Problem ist, wann offer()false zurückgegeben werden sollte. Die aktuell akzeptierte Antwort gibt false zurück, wenn die Warteschlange einige Aufgaben enthält. Wie ich in meinem Kommentar dort ausgeführt habe, führt dies jedoch zu unerwünschten Effekten. Wenn Sie alternativ immer false zurückgeben, werden auch dann immer wieder neue Threads erzeugt, wenn Threads in der Warteschlange warten.

Die Lösung besteht darin, Java 7 zu verwenden LinkedTransferQueueund einen offer()Anruf zu tätigen tryTransfer(). Wenn ein Consumer-Thread wartet, wird die Aufgabe nur an diesen Thread übergeben. Andernfalls offer()wird false zurückgegeben und ThreadPoolExecutorein neuer Thread erzeugt.

    BlockingQueue<Runnable> queue = new LinkedTransferQueue<Runnable>() {
        @Override
        public boolean offer(Runnable e) {
            return tryTransfer(e);
        }
    };
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 50, 60, TimeUnit.SECONDS, queue);
    threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            try {
                executor.getQueue().put(r);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    });
Robert Tupelo-Schneck
quelle
Ich muss zustimmen, das sieht für mich am saubersten aus. Der einzige Nachteil der Lösung ist, dass LinkedTransferQueue unbegrenzt ist, sodass Sie ohne zusätzliche Arbeit keine kapazitätsgebundene Aufgabenwarteschlange erhalten.
Yeroc
Es gibt ein Problem, wenn der Pool die maximale Größe erreicht. Angenommen, der Pool ist auf die maximale Größe skaliert und jeder Thread führt derzeit eine Aufgabe aus. Wenn ausführbar gesendet wird, gibt dieses Angebot impl false zurück und ThreadPoolExecutor versucht, einen Worker-Thread hinzuzufügen, aber der Pool hat bereits sein Maximum erreicht, sodass ausführbar einfach abgelehnt wird. Gemäß dem von Ihnen geschriebenen abgelehnten ExceHandler wird es erneut in die Warteschlange gestellt, was dazu führt, dass dieser Affentanz von Anfang an wieder stattfindet.
Sudheera
1
@ Sudheera Ich glaube du liegst falsch. queue.offer(), weil es tatsächlich aufruft LinkedTransferQueue.tryTransfer(), wird false zurückgeben und die Aufgabe nicht in die Warteschlange stellen. Allerdings die RejectedExecutionHandlerAufrufe queue.put(), die nicht fehlschlagen und die Aufgabe in die Warteschlange stellen.
Robert Tupelo-Schneck
1
@ RobertTupelo-Schneck äußerst nützlich und nett!
Eugene
1
@ RobertTupelo-Schneck Funktioniert wie ein Zauber! Ich weiß nicht, warum es in Java nicht so etwas aus der Box gibt
Georgi Peev
7

Hinweis: Ich bevorzuge und empfehle jetzt meine andere Antwort .

Hier ist eine Version, die sich für mich viel einfacher anfühlt: Erhöhen Sie die CorePoolSize (bis zur Grenze von MaximumPoolSize), wenn eine neue Aufgabe ausgeführt wird, und verringern Sie dann die CorePoolSize (bis zur Grenze der vom Benutzer angegebenen "Core-Pool-Größe"), wann immer a Aufgabe abgeschlossen.

Um es anders auszudrücken, verfolgen Sie die Anzahl der ausgeführten oder in die Warteschlange gestellten Aufgaben und stellen Sie sicher, dass corePoolSize der Anzahl der Aufgaben entspricht, solange sie zwischen der vom Benutzer angegebenen "Kernpoolgröße" und der maximalenPoolSize liegt.

public class GrowBeforeQueueThreadPoolExecutor extends ThreadPoolExecutor {
    private int userSpecifiedCorePoolSize;
    private int taskCount;

    public GrowBeforeQueueThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        userSpecifiedCorePoolSize = corePoolSize;
    }

    @Override
    public void execute(Runnable runnable) {
        synchronized (this) {
            taskCount++;
            setCorePoolSizeToTaskCountWithinBounds();
        }
        super.execute(runnable);
    }

    @Override
    protected void afterExecute(Runnable runnable, Throwable throwable) {
        super.afterExecute(runnable, throwable);
        synchronized (this) {
            taskCount--;
            setCorePoolSizeToTaskCountWithinBounds();
        }
    }

    private void setCorePoolSizeToTaskCountWithinBounds() {
        int threads = taskCount;
        if (threads < userSpecifiedCorePoolSize) threads = userSpecifiedCorePoolSize;
        if (threads > getMaximumPoolSize()) threads = getMaximumPoolSize();
        setCorePoolSize(threads);
    }
}

Wie geschrieben, unterstützt die Klasse das Ändern der benutzerdefinierten CorePoolSize oder MaximumPoolSize nach der Erstellung nicht und das Manipulieren der Arbeitswarteschlange nicht direkt oder über remove()oder purge().

Robert Tupelo-Schneck
quelle
Ich mag es bis auf die synchronizedBlöcke. Können Sie die Warteschlange anrufen, um die Anzahl der Aufgaben zu ermitteln? Oder vielleicht eine verwenden AtomicInteger?
Grau
Ich wollte sie vermeiden, aber das Problem ist das hier. Wenn es eine Anzahl von execute()Anrufen in separaten Threads gibt, wird jeder (1) herausfinden, wie viele Threads benötigt werden, (2) setCorePoolSizediese Nummer und (3) anrufen super.execute(). Wenn die Schritte (1) und (2) nicht synchronisiert sind, bin ich mir nicht sicher, wie ich eine unglückliche Reihenfolge verhindern kann, bei der Sie die Größe des Kernpools nach einer höheren Zahl auf eine niedrigere Zahl setzen. Mit direktem Zugriff auf das Feld der Oberklasse könnte dies stattdessen mit compare-and-set erfolgen, aber ich sehe keinen sauberen Weg, dies in einer Unterklasse ohne Synchronisation zu tun.
Robert Tupelo-Schneck
Ich denke, die Strafen für diese Rennbedingung sind relativ niedrig, solange das taskCountFeld gültig ist (dh a AtomicInteger). Wenn zwei Threads die Poolgröße unmittelbar nacheinander neu berechnen, sollten die richtigen Werte erhalten werden. Wenn der zweite die Kern-Threads verkleinert, muss er einen Abfall in der Warteschlange oder so gesehen haben.
Grau
1
Leider denke ich, dass es schlimmer ist. Angenommen, die Aufgaben 10 und 11 werden aufgerufen execute(). Jeder wird anrufen atomicTaskCount.incrementAndGet()und 10 bzw. 11 erhalten. Aber ohne Synchronisation (über das Abrufen der Aufgabenanzahl und das Festlegen der Kernpoolgröße) könnten Sie dann erhalten, dass (1) Aufgabe 11 die Kernpoolgröße auf 11 setzt, (2) Aufgabe 10 die Kernpoolgröße auf 10 setzt, (3) Task 10 ruft auf super.execute(), (4) Task 11 ruft auf super.execute()und wird in die Warteschlange gestellt.
Robert Tupelo-Schneck
2
Ich habe diese Lösung ernsthaft getestet und sie ist eindeutig die beste. In einer Umgebung mit starkem Multithreading wird es manchmal immer noch in die Warteschlange gestellt, wenn freie Threads vorhanden sind (aufgrund der TPE.execute-Natur mit freiem Thread), aber es kommt selten vor, im Gegensatz zu der als Antwort gekennzeichneten Lösung, bei der die Race-Bedingungen mehr Chancen haben passieren, so passiert dies so ziemlich bei jedem Multithread-Lauf.
Willst du alles wissen
6

Wir haben eine Unterklasse ThreadPoolExecutor, die eine zusätzliche nimmt creationThresholdund überschreibt execute.

public void execute(Runnable command) {
    super.execute(command);
    final int poolSize = getPoolSize();
    if (poolSize < getMaximumPoolSize()) {
        if (getQueue().size() > creationThreshold) {
            synchronized (this) {
                setCorePoolSize(poolSize + 1);
                setCorePoolSize(poolSize);
            }
        }
    }
}

Vielleicht hilft das auch, aber deine sieht natürlich künstlerischer aus ...

Ralf H.
quelle
Interessant. Danke dafür. Ich wusste eigentlich nicht, dass die Kerngröße veränderlich ist.
Grau
Jetzt, wo ich etwas mehr darüber nachdenke, ist diese Lösung besser als meine, wenn es darum geht, die Größe der Warteschlange zu überprüfen. Ich habe meine Antwort so angepasst , dass die offer(...)Methode nur falsebedingt zurückgegeben wird. Vielen Dank!
Grau
4

Die empfohlene Antwort behebt nur eines (1) des Problems mit dem JDK-Thread-Pool:

  1. JDK-Thread-Pools sind auf Warteschlangen ausgerichtet. Anstatt einen neuen Thread zu erzeugen, werden sie die Aufgabe in die Warteschlange stellen. Nur wenn die Warteschlange ihr Limit erreicht, erzeugt der Thread-Pool einen neuen Thread.

  2. Das Zurückziehen des Threads erfolgt nicht, wenn die Last leichter wird. Wenn beispielsweise eine Reihe von Jobs auf den Pool trifft, die dazu führen, dass der Pool maximal wird, gefolgt von einer geringen Last von maximal 2 Aufgaben gleichzeitig, verwendet der Pool alle Threads, um die geringe Last zu bedienen, wodurch ein Zurückziehen des Threads verhindert wird. (nur 2 Threads würden benötigt ...)

Unzufrieden mit dem oben genannten Verhalten habe ich einen Pool implementiert, um die oben genannten Mängel zu beheben.

So beheben Sie das Problem: 2) Die Verwendung der Lifo-Zeitplanung behebt das Problem. Diese Idee wurde von Ben Maurer auf der ACM-Konferenz 2015 vorgestellt: Systems @ Facebook scale

So wurde eine neue Implementierung geboren:

LifoThreadPoolExecutorSQP

Bisher verbessert diese Implementierung die Leistung der asynchronen Ausführung für ZEL .

Die Implementierung ist spinfähig, um den Overhead für Kontextwechsel zu reduzieren und für bestimmte Anwendungsfälle eine überlegene Leistung zu erzielen.

Ich hoffe es hilft...

PS: JDK Fork Join Pool implementiert ExecutorService und arbeitet als "normaler" Thread-Pool. Die Implementierung ist performant. Sie verwendet die LIFO-Thread-Planung. Es gibt jedoch keine Kontrolle über die Größe der internen Warteschlange, das Zeitlimit für den Ruhestand ... und vor allem können Aufgaben nicht ausgeführt werden beim Abbrechen unterbrochen

user2179737
quelle
1
Schade, dass diese Implementierung so viele externe Abhängigkeiten hat. Für mich nutzlos machen: - /
Martin L.
1
Es ist ein wirklich guter Punkt (2.). Leider ist die Implementierung nicht aus externen Abhängigkeiten ersichtlich, kann aber dennoch übernommen werden, wenn Sie möchten.
Alexey Vlasov
1

Hinweis: Ich bevorzuge und empfehle jetzt meine andere Antwort .

Ich habe einen anderen Vorschlag, der der ursprünglichen Idee folgt, die Warteschlange so zu ändern, dass sie false zurückgibt. In diesem Fall können alle Aufgaben in die Warteschlange aufgenommen werden. Wenn jedoch eine Aufgabe danach in die Warteschlange gestellt wird, execute()folgt eine Sentinel-No-Op-Aufgabe, die von der Warteschlange abgelehnt wird, wodurch ein neuer Thread erzeugt wird, der das No-Op unmittelbar danach ausführt etwas aus der Warteschlange.

Da Worker-Threads möglicherweise LinkedBlockingQueuenach einer neuen Aufgabe abfragen, kann eine Aufgabe auch dann in die Warteschlange gestellt werden, wenn ein Thread verfügbar ist. Um zu vermeiden, dass neue Threads erzeugt werden, selbst wenn Threads verfügbar sind, müssen wir verfolgen, wie viele Threads auf neue Aufgaben in der Warteschlange warten, und einen neuen Thread nur dann erzeugen, wenn sich mehr Aufgaben in der Warteschlange befinden als wartende Threads.

final Runnable SENTINEL_NO_OP = new Runnable() { public void run() { } };

final AtomicInteger waitingThreads = new AtomicInteger(0);

BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {
    @Override
    public boolean offer(Runnable e) {
        // offer returning false will cause the executor to spawn a new thread
        if (e == SENTINEL_NO_OP) return size() <= waitingThreads.get();
        else return super.offer(e);
    }

    @Override
    public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
        try {
            waitingThreads.incrementAndGet();
            return super.poll(timeout, unit);
        } finally {
            waitingThreads.decrementAndGet();
        }
    }

    @Override
    public Runnable take() throws InterruptedException {
        try {
            waitingThreads.incrementAndGet();
            return super.take();
        } finally {
            waitingThreads.decrementAndGet();
        }
    }
};

ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 50, 60, TimeUnit.SECONDS, queue) {
    @Override
    public void execute(Runnable command) {
        super.execute(command);
        if (getQueue().size() > waitingThreads.get()) super.execute(SENTINEL_NO_OP);
    }
};
threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (r == SENTINEL_NO_OP) return;
        else throw new RejectedExecutionException();            
    }
});
Robert Tupelo-Schneck
quelle
0

Die beste Lösung, die ich mir vorstellen kann, ist zu erweitern.

ThreadPoolExecutorbietet einige Hakenmethoden an: beforeExecuteund afterExecute. In Ihrer Erweiterung können Sie eine begrenzte Warteschlange zum Einspeisen von Aufgaben und eine zweite unbegrenzte Warteschlange zum Behandeln des Überlaufs verwenden. Wenn jemand anruft submit, können Sie versuchen, die Anforderung in die begrenzte Warteschlange zu stellen. Wenn Sie auf eine Ausnahme stoßen, stecken Sie die Aufgabe einfach in Ihre Überlaufwarteschlange. Sie können dann den afterExecuteHook verwenden, um festzustellen, ob sich nach Abschluss einer Aufgabe etwas in der Überlaufwarteschlange befindet. Auf diese Weise kümmert sich der Executor zuerst um das Material in seiner begrenzten Warteschlange und zieht automatisch aus dieser unbegrenzten Warteschlange, wenn es die Zeit erlaubt.

Es scheint mehr Arbeit als Ihre Lösung zu sein, aber es geht zumindest nicht darum, Warteschlangen unerwartetes Verhalten zu geben. Ich stelle mir auch vor, dass es eine bessere Möglichkeit gibt, den Status der Warteschlange und der Threads zu überprüfen, als sich auf Ausnahmen zu verlassen, die relativ langsam ausgelöst werden.

bstempi
quelle
Ich mag diese Lösung nicht. Ich bin mir ziemlich sicher, dass ThreadPoolExecutor nicht für die Vererbung entwickelt wurde.
Scottb
Es gibt tatsächlich ein Beispiel für eine Erweiterung direkt im JavaDoc. Sie geben an, dass die meisten wahrscheinlich nur die Hook-Methoden implementieren werden, aber sie sagen Ihnen, worauf Sie beim Erweitern noch achten müssen.
Bstempi
0

Hinweis: Wenn Sie für JDK ThreadPoolExecutor eine begrenzte Warteschlange haben, erstellen Sie nur dann neue Threads, wenn das Angebot false zurückgibt . Mit CallerRunsPolicy erhalten Sie möglicherweise etwas Nützliches , das ein wenig BackPressure erzeugt und Aufrufe direkt im Aufrufer-Thread ausführt.

Ich brauche Aufgaben aus Fäden am Pool und haben eine ubounded Warteschlange für die Planung erstellt ausgeführt werden, während die Anzahl der Threads im Pool können wachsen oder schrumpfen zwischen corePoolSize und maximumPoolSize so ...

Am Ende habe ich ein vollständiges Kopieren und Einfügen aus ThreadPoolExecutor durchgeführt und die Ausführungsmethode ein wenig geändert , da dies leider nicht durch Erweiterung möglich war (es werden private Methoden aufgerufen ).

Ich wollte nicht sofort neue Threads erzeugen, wenn eine neue Anfrage eintrifft und alle Threads beschäftigt sind (weil ich im Allgemeinen kurzlebige Aufgaben habe). Ich habe einen Schwellenwert hinzugefügt, kann ihn aber jederzeit an Ihre Bedürfnisse anpassen (möglicherweise ist es für IO meistens besser, diesen Schwellenwert zu entfernen).

private final AtomicInteger activeWorkers = new AtomicInteger(0);
private volatile double threshold = 0.7d;

protected void beforeExecute(Thread t, Runnable r) {
    activeWorkers.incrementAndGet();
}
protected void afterExecute(Runnable r, Throwable t) {
    activeWorkers.decrementAndGet();
}
public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();

        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }

        if (isRunning(c) && this.workQueue.offer(command)) {
            int recheck = this.ctl.get();
            if (!isRunning(recheck) && this.remove(command)) {
                this.reject(command);
            } else if (workerCountOf(recheck) == 0) {
                this.addWorker((Runnable) null, false);
            }
            //>>change start
            else if (workerCountOf(recheck) < maximumPoolSize //
                && (activeWorkers.get() > workerCountOf(recheck) * threshold
                    || workQueue.size() > workerCountOf(recheck) * threshold)) {
                this.addWorker((Runnable) null, false);
            }
            //<<change end
        } else if (!this.addWorker(command, false)) {
            this.reject(command);
        }
    }
Radu Toader
quelle