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

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.

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.

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

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

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

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

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

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

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.

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