Angepasster Thread-Pool in parallelem Java 8-Stream

Ist es möglich, zum angeben einer benutzerdefinierten thread-pool für Java 8 parallel stream? Ich kann es nirgends finden.

Vorstellen, dass ich eine server-Anwendung, und ich möchte die parallele streams. Aber die Anwendung ist groß und multi-threaded, so will ich es aufteilen. Ich will nicht einen langsam Laufenden Aufgabe in einem Modul der applicationblock Aufgaben von einem anderen Modul.

Wenn ich nicht verwenden können, die verschiedenen thread-pools für die verschiedenen Module, das bedeutet, ich kann sicher nicht die parallele streams in den meisten realen Situationen.

Versuchen Sie das folgende Beispiel. Es gibt einige CPU-intensive Aufgaben ausgeführt, die in separaten threads.
Die Aufgaben nutzen die parallele streams. Die erste Aufgabe ist gebrochen, so dass jeder Schritt dauert 1 Sekunde (simuliert durch den thread schlafen). Das Problem ist, dass andere threads stecken und warten, bis die defekte Aufgabe zu beenden. Das ist erfundenes Beispiel, aber stellen Sie sich ein servlet-app und jemand der Einreichung eine lange laufende Aufgabe der gemeinsamen fork-join-pool.

public class ParallelTest {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService es = Executors.newCachedThreadPool();

        es.execute(() -> runTask(1000)); //incorrect task
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));


        es.shutdown();
        es.awaitTermination(60, TimeUnit.SECONDS);
    }

    private static void runTask(int delay) {
        range(1, 1_000_000).parallel().filter(ParallelTest::isPrime).peek(i -> Utils.sleep(delay)).max()
                .ifPresent(max -> System.out.println(Thread.currentThread() + " " + max));
    }

    public static boolean isPrime(long n) {
        return n > 1 && rangeClosed(2, (long) sqrt(n)).noneMatch(divisor -> n % divisor == 0);
    }
}
Kommentar zu dem Problem
Was meinst du mit custom thread-pool? Es gibt einen einzigen gemeinsamen ForkJoinPool, aber Sie können immer Ihre eigenen erstellen ForkJoinPool und Anforderungen übermitteln. Kommentarautor: edharned
Hinweis: Java Champion Heinz Kabutz inspiziert das gleiche problem, aber mit noch schlimmeren Auswirkungen: Deadlocks threads der gemeinsamen fork-join-pool. Siehe javaspecialists.eu/archive/Issue223.html Kommentarautor: Peti

InformationsquelleAutor der Frage Lukas | 2014-01-16

  1. 279

    Gibt es eigentlich einen trick, wie das ausführen einer parallelen Betrieb in einem bestimmten fork-join-pool. Wenn Sie ausführen, es als eine Aufgabe in einem fork-join-pool, bleibt es dort und nicht die häufigste. 

    ForkJoinPool forkJoinPool = new ForkJoinPool(2);
forkJoinPool.submit(() ->
    //parallel task here, for example
    IntStream.range(1, 1_000_000).parallel().filter(PrimesPrint::isPrime).collect(toList())
).get();

    Der trick basiert auf ForkJoinTask.Gabel, die angibt: "Ordnet asynchron ausführen dieser Aufgabe in den pool der aktuellen Aufgabe ausgeführt wird, falls anwendbar, oder über den ForkJoinPool.commonPool (), wenn nicht inForkJoinPool()"

    InformationsquelleAutor der Antwort Lukas

  2. 138

    Den parallelen streams verwenden Sie die Standard - ForkJoinPool.commonPool die standardmäßig hat man weniger threads, wie Sie Prozessoren, zurückgegeben durch Runtime.getRuntime().availableProcessors() (Das bedeutet, dass parallele streams verwenden Sie alle Ihre Prozessoren, da Sie auch mit dem Haupt-thread):

    Für Anwendungen, die einen separaten oder benutzerdefinierte pools, einen ForkJoinPool können konstruiert werden mit einem bestimmten Ziel Parallelität Ebene; standardmäßig gleich der Anzahl der verfügbaren Prozessoren.

    Dies bedeutet auch, wenn Sie geschachtelte parallel-streams oder mehrere parallele streams gleichzeitig begann, Sie alle teilen dem gleichen pool. Vorteil: Sie werden nie mehr als die Vorgabe (Anzahl der verfügbaren Prozessoren). Nachteil: man kann nicht "alle Prozessoren" zugeordnet, um jeden parallelen stream zu initiieren (wenn Sie geschehen, um mehr als eine). (Anscheinend können Sie ein ManagedBlocker zu umgehen.)

    Verändern die Art und Weise parallele streams ausgeführt werden, können Sie entweder

    • reichen die parallelen stream-Ausführung, um Ihren eigenen ForkJoinPool: yourFJP.submit(() -> stream.parallel().forEach(soSomething)).get(); oder
    • Sie können ändern Sie die Größe des gemeinsam genutzten pool mit system Eigenschaften: System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20") für ein Ziel Parallelität von 20 threads.

    Beispiel des letzteren auf meiner Maschine, die hat 8 Prozessoren. Wenn ich das folgende Programm ausführen:

    long start = System.currentTimeMillis();
IntStream s = IntStream.range(0, 20);
//System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");
s.parallel().forEach(i -> {
    try { Thread.sleep(100); } catch (Exception ignore) {}
    System.out.print((System.currentTimeMillis() - start) + " ");
});

    Ausgabe:

    215 216 216 216 216 216 216 216 315 316 316 316 316 316 316 316 415 416 416 416

    Damit Sie sehen können, dass die parallele stream-Prozesse 8 items gleichzeitig, D. H. es nutzt 8 threads. Wenn ich allerdings kommentieren Sie die auskommentierte Zeile, die Ausgabe ist:

    215 215 215 215 215 216 216 216 216 216 216 216 216 216 216 216 216 216 216 216

    Dieser Zeit, der parallelen Streams benutzt hat 20 threads und alle 20 Elemente in dem stream wurden gleichzeitig verarbeitet.

    InformationsquelleAutor der Antwort assylias

  3. 31

    Alternativ zu dem trick mit der Auslösung der parallelen Berechnung in Ihrem eigenen forkJoinPool Sie kann auch passieren, dass der pool der CompletableFuture.supplyAsync Methode, wie in:

    ForkJoinPool forkJoinPool = new ForkJoinPool(2);
CompletableFuture<List<Integer>> primes = CompletableFuture.supplyAsync(() ->
    //parallel task here, for example
    range(1, 1_000_000).parallel().filter(PrimesPrint::isPrime).collect(toList()), 
    forkJoinPool
);

    InformationsquelleAutor der Antwort Mario Fusco

  4. 11

    Mit einem ForkJoinPool und reichen für einen parallelen stream nicht zuverlässig nutzen alle threads. Wenn man sich diese ( Parallel stream von einem HashSet nicht parallel laufen ) und diese ( Warum wird der parallele Datenstrom nicht alle threads von den ForkJoinPool? ), werden Sie sehen, die Argumentation.

    Kurze version: wenn ForkJoinPool/reichen nicht für Sie arbeiten, verwenden Sie 

    System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "10");

    InformationsquelleAutor der Antwort Tod Casasent

  5. 6

    Bis jetzt habe ich die beschriebenen Lösungen in den Antworten dieser Frage. Nun, ich kam mit einer kleinen Bibliothek namens Parallel-Stream-Unterstützung:

    ForkJoinPool pool = new ForkJoinPool(NR_OF_THREADS);
ParallelIntStreamSupport.range(1, 1_000_000, pool)
    .filter(PrimesPrint::isPrime)
    .collect(toList())

    Aber wie @PabloMatiasGomez darauf hingewiesen, in den Kommentaren, gibt es Nachteile bezüglich der Trennung Mechanismus der parallelen Ströme, die hängt stark von der Größe der gemeinsamen pool. Sehen Parallel stream von einem HashSet nicht parallel laufen .

    Bin ich mit dieser Lösung nur über separate pools für unterschiedliche Arten von Arbeit, aber ich kann nicht die Größe des common-pool 1, auch wenn ich Sie nicht nutze.

    InformationsquelleAutor der Antwort Stefan Ferstl

  6. 5

    Zur Messung der tatsächlichen Anzahl der verwendeten threads, Sie können überprüfen Thread.activeCount():

        Runnable r = () -> IntStream
            .range(-42, +42)
            .parallel()
            .map(i -> Thread.activeCount())
            .max()
            .ifPresent(System.out::println);

    ForkJoinPool.commonPool().submit(r).join();
    new ForkJoinPool(42).submit(r).join();

    Diese produzieren können auf einen 4-Kern-CPU eine Ausgabe wie:

    5 //common pool
23 //custom pool

    Ohne .parallel() es gibt:

    3 //common pool
4 //custom pool

    InformationsquelleAutor der Antwort charlie

  7. 1

    Gehen, um AbacusUtil. Thread-Anzahl kann durch die angegebene, für die parallele Streams. Hier ist der Beispielcode:

    LongStream.range(4, 1_000_000).parallel(threadNum)...

    Offenlegung： ich bin der Entwickler des AbacusUtil.

    InformationsquelleAutor der Antwort user_3380739

  8. 0

    Wenn Sie nichts dagegen haben, mit einem Drittanbieter-Bibliothek, mit cyclops-reagieren können Sie mischen sequentielle und parallele Streams in der gleichen Rohrleitung und bieten maßgeschneiderte ForkJoinPools. Zum Beispiel

     ReactiveSeq.range(1, 1_000_000)
            .foldParallel(new ForkJoinPool(10),
                          s->s.filter(i->true)
                              .peek(i->System.out.println("Thread " + Thread.currentThread().getId()))
                              .max(Comparator.naturalOrder()));

    Oder wenn wir wollten weiterhin die Bearbeitung innerhalb einer sequentiellen Stream

     ReactiveSeq.range(1, 1_000_000)
            .parallel(new ForkJoinPool(10),
                      s->s.filter(i->true)
                          .peek(i->System.out.println("Thread " + Thread.currentThread().getId())))
            .map(this::processSequentially)
            .forEach(System.out::println);

    [Offenlegung ich bin der leitende Entwickler des Zyklopen-reagieren]

    InformationsquelleAutor der Antwort John McClean

Schreibe einen Kommentar