Kumpulan utas khusus dalam aliran paralel Java 8

Apakah mungkin untuk menentukan kumpulan utas khusus untuk aliran paralel Java 8 ? Saya tidak dapat menemukannya di mana pun.

Bayangkan saya memiliki aplikasi server dan saya ingin menggunakan aliran paralel. Tetapi aplikasinya besar dan multi-utas jadi saya ingin mengelompokkannya. Saya tidak ingin tugas berjalan lambat dalam satu modul tugas aplikasiblock dari modul lain.

Jika saya tidak dapat menggunakan kumpulan utas yang berbeda untuk modul yang berbeda, itu berarti saya tidak dapat menggunakan aliran paralel secara aman di sebagian besar situasi dunia nyata.

Coba contoh berikut ini. Ada beberapa tugas intensif CPU yang dijalankan dalam utas terpisah. Tugas memanfaatkan aliran paralel. Tugas pertama rusak, sehingga setiap langkah membutuhkan waktu 1 detik (disimulasikan oleh thread sleep). Masalahnya adalah bahwa utas lainnya macet dan menunggu tugas yang rusak selesai. Ini adalah contoh yang dibuat-buat, tetapi bayangkan aplikasi servlet dan seseorang yang mengirimkan tugas yang sudah berjalan lama ke pool bersama garpu gabungan.

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

Sebenarnya ada trik bagaimana menjalankan operasi paralel di kumpulan fork-join tertentu. Jika Anda menjalankannya sebagai tugas di kumpulan garpu-bergabung, itu tetap di sana dan tidak menggunakan yang umum.

final int parallelism = 4;
ForkJoinPool forkJoinPool = null;
try {
    forkJoinPool = new ForkJoinPool(parallelism);
    final List<Integer> primes = forkJoinPool.submit(() ->
        // Parallel task here, for example
        IntStream.range(1, 1_000_000).parallel()
                .filter(PrimesPrint::isPrime)
                .boxed().collect(Collectors.toList())
    ).get();
    System.out.println(primes);
} catch (InterruptedException | ExecutionException e) {
    throw new RuntimeException(e);
} finally {
    if (forkJoinPool != null) {
        forkJoinPool.shutdown();
    }
}

Trik ini didasarkan pada ForkJoinTask.fork yang menetapkan: "Mengatur untuk menjalankan tugas ini secara serempak di kumpulan tugas yang sedang berjalan, jika ada, atau menggunakan ForkJoinPool.commonPool () jika tidak diForkJoinPool ()"

Aliran paralel menggunakan default ForkJoinPool.commonPoolyang secara default memiliki satu utas lebih sedikit saat Anda memiliki prosesor , seperti yang dikembalikan oleh Runtime.getRuntime().availableProcessors()(Ini berarti aliran paralel menggunakan semua prosesor Anda karena mereka juga menggunakan utas utama):

Untuk aplikasi yang memerlukan kumpulan terpisah atau khusus, ForkJoinPool dapat dibangun dengan tingkat paralelisme target yang diberikan; secara default, sama dengan jumlah prosesor yang tersedia.

Ini juga berarti jika Anda memiliki stream paralel bersarang atau beberapa stream paralel mulai bersamaan, mereka semua akan berbagi kumpulan yang sama. Keuntungan: Anda tidak akan pernah menggunakan lebih dari standar (jumlah prosesor yang tersedia). Kerugian: Anda mungkin tidak mendapatkan "semua prosesor" yang ditetapkan untuk setiap aliran paralel yang Anda mulai (jika Anda memiliki lebih dari satu). (Rupanya Anda dapat menggunakan ManagedBlocker untuk mengelak dari itu.)

Untuk mengubah cara aliran paralel dieksekusi, Anda bisa

• kirimkan eksekusi aliran paralel ke ForkJoinPool Anda sendiri: yourFJP.submit(() -> stream.parallel().forEach(soSomething)).get();atau
• Anda dapat mengubah ukuran kumpulan umum menggunakan properti sistem: System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20")untuk paralelisme target 20 utas. Namun, ini tidak lagi berfungsi setelah tambalan backported https://bugs.openjdk.java.net/browse/JDK-8190974 .

Contoh yang terakhir pada mesin saya yang memiliki 8 prosesor. Jika saya menjalankan program berikut:

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

Outputnya adalah:

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

Jadi, Anda dapat melihat bahwa aliran paralel memproses 8 item sekaligus, yaitu menggunakan 8 utas. Namun, jika saya batalkan komentar pada baris komentar, hasilnya adalah:

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

Kali ini, aliran paralel telah menggunakan 20 utas dan semua 20 elemen dalam aliran telah diproses secara bersamaan.

Atau dengan trik memicu komputasi paralel di dalam forkJoinPool Anda sendiri, Anda juga dapat meneruskan kumpulan itu ke metode CompletableFuture.supplyAsync seperti di:

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

Solusi asli (pengaturan properti paralelisme umum ForkJoinPool) tidak lagi berfungsi. Melihat tautan dalam jawaban asli, pembaruan yang memecah ini telah kembali porting ke Java 8. Seperti yang disebutkan dalam utas yang tertaut, solusi ini tidak dijamin berfungsi selamanya. Berdasarkan itu, solusinya adalah forkjoinpool.submit dengan .get solusi yang dibahas dalam jawaban yang diterima. Saya pikir backport memperbaiki tidak dapat diandalkannya solusi ini juga.

ForkJoinPool fjpool = new ForkJoinPool(10);
System.out.println("stream.parallel");
IntStream range = IntStream.range(0, 20);
fjpool.submit(() -> range.parallel()
        .forEach((int theInt) ->
        {
            try { Thread.sleep(100); } catch (Exception ignore) {}
            System.out.println(Thread.currentThread().getName() + " -- " + theInt);
        })).get();
System.out.println("list.parallelStream");
int [] array = IntStream.range(0, 20).toArray();
List<Integer> list = new ArrayList<>();
for (int theInt: array)
{
    list.add(theInt);
}
fjpool.submit(() -> list.parallelStream()
        .forEach((theInt) ->
        {
            try { Thread.sleep(100); } catch (Exception ignore) {}
            System.out.println(Thread.currentThread().getName() + " -- " + theInt);
        })).get();

Kami dapat mengubah paralelisme default menggunakan properti berikut:

-Djava.util.concurrent.ForkJoinPool.common.parallelism=16

yang dapat mengatur untuk menggunakan lebih banyak paralelisme.

Untuk mengukur jumlah utas yang sebenarnya digunakan, Anda dapat memeriksa 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();

Ini dapat menghasilkan output CPU 4-core seperti:

5 // common pool
23 // custom pool

Tanpa .parallel()itu memberi:

3 // common pool
4 // custom pool

Sampai sekarang, saya menggunakan solusi yang dijelaskan dalam jawaban dari pertanyaan ini. Sekarang, saya datang dengan perpustakaan kecil bernama Parallel Stream Support untuk itu:

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

Tetapi seperti yang ditunjukkan oleh @PabloMatiasGomez dalam komentar, ada kelemahan terkait mekanisme pemisahan aliran paralel yang sangat bergantung pada ukuran kumpulan umum. Lihat Aliran paralel dari HashSet tidak berjalan secara paralel .

Saya menggunakan solusi ini hanya untuk memiliki kolam yang terpisah untuk berbagai jenis pekerjaan tetapi saya tidak dapat mengatur ukuran kolam yang umum menjadi 1 bahkan jika saya tidak menggunakannya.

Catatan: Tampaknya ada perbaikan yang diterapkan di JDK 10 yang memastikan Custom Thread Pool menggunakan jumlah utas yang diharapkan.

Eksekusi aliran paralel dalam ForkJoinPool khusus harus mematuhi paralelisme https://bugs.openjdk.java.net/browse/JDK-8190974

Saya mencoba ForkJoinPool kustom sebagai berikut untuk menyesuaikan ukuran kolam:

private static Set<String> ThreadNameSet = new HashSet<>();
private static Callable<Long> getSum() {
    List<Long> aList = LongStream.rangeClosed(0, 10_000_000).boxed().collect(Collectors.toList());
    return () -> aList.parallelStream()
            .peek((i) -> {
                String threadName = Thread.currentThread().getName();
                ThreadNameSet.add(threadName);
            })
            .reduce(0L, Long::sum);
}

private static void testForkJoinPool() {
    final int parallelism = 10;

    ForkJoinPool forkJoinPool = null;
    Long result = 0L;
    try {
        forkJoinPool = new ForkJoinPool(parallelism);
        result = forkJoinPool.submit(getSum()).get(); //this makes it an overall blocking call

    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    } finally {
        if (forkJoinPool != null) {
            forkJoinPool.shutdown(); //always remember to shutdown the pool
        }
    }
    out.println(result);
    out.println(ThreadNameSet);
}

Ini adalah output yang mengatakan bahwa pool menggunakan lebih banyak thread daripada default 4 .

50000005000000
[ForkJoinPool-1-worker-8, ForkJoinPool-1-worker-9, ForkJoinPool-1-worker-6, ForkJoinPool-1-worker-11, ForkJoinPool-1-worker-10, ForkJoinPool-1-worker-1, ForkJoinPool-1-worker-15, ForkJoinPool-1-worker-13, ForkJoinPool-1-worker-4, ForkJoinPool-1-worker-2]

Tetapi sebenarnya ada yang aneh , ketika saya mencoba untuk mencapai hasil yang sama menggunakan ThreadPoolExecutorsebagai berikut:

BlockingDeque blockingDeque = new LinkedBlockingDeque(1000);
ThreadPoolExecutor fixedSizePool = new ThreadPoolExecutor(10, 20, 60, TimeUnit.SECONDS, blockingDeque, new MyThreadFactory("my-thread"));

tapi saya gagal.

Ini hanya akan memulai parallelStream di utas baru dan kemudian yang lainnya sama, yang sekali lagi membuktikan bahwa parallelStreamakan menggunakan ForkJoinPool untuk memulai utas anaknya.

Pergi untuk mendapatkan AbacusUtil . Nomor utas dapat ditentukan untuk aliran paralel. Berikut ini contoh kode:

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

Pengungkapan ： Saya pengembang AbacusUtil.

Jika Anda tidak ingin mengandalkan hacks implementasi, selalu ada cara untuk mencapai hal yang sama dengan menerapkan kolektor khusus yang akan menggabungkan mapdan collectsemantik ... dan Anda tidak akan terbatas pada ForkJoinPool:

list.stream()
  .collect(parallelToList(i -> fetchFromDb(i), executor))
  .join()

Untungnya, sudah selesai di sini dan tersedia di Maven Central: http://github.com/pivovarit/parallel-collectors

Penafian: Saya menulisnya dan bertanggung jawab untuk itu.

Jika Anda tidak keberatan menggunakan perpustakaan pihak ketiga, dengan cyclops-react Anda dapat mencampur Streaming berurutan dan paralel dalam pipa yang sama dan memberikan ForkJoinPools khusus. Sebagai contoh

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

Atau jika kami ingin melanjutkan pemrosesan dalam aliran berurutan

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

[Pengungkapan Saya adalah pengembang utama cyclops-react]

Jika Anda tidak memerlukan ThreadPool khusus tetapi Anda ingin membatasi jumlah tugas bersamaan, Anda dapat menggunakan:

List<Path> paths = List.of("/path/file1.csv", "/path/file2.csv", "/path/file3.csv").stream().map(e -> Paths.get(e)).collect(toList());
List<List<Path>> partitions = Lists.partition(paths, 4); // Guava method

partitions.forEach(group -> group.parallelStream().forEach(csvFilePath -> {
       // do your processing   
}));

(Pertanyaan rangkap yang meminta ini dikunci, jadi tolong bawa saya ke sini)

Anda dapat mencoba mengimplementasikan ForkJoinWorkerThreadFactory ini dan menyuntikkannya ke kelas Fork-Join.

public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode) {
        this(checkParallelism(parallelism),
             checkFactory(factory),
             handler,
             asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
             "ForkJoinPool-" + nextPoolId() + "-worker-");
        checkPermission();
    }

Anda dapat menggunakan konstruktor kolam Fork-Join ini untuk melakukan ini.

Catatan: - 1. jika Anda menggunakan ini, pertimbangkan bahwa berdasarkan implementasi Anda dari thread baru, penjadwalan dari JVM akan terpengaruh, yang umumnya menjadwalkan fork-join threads ke core yang berbeda (diperlakukan sebagai utas komputasi). 2. penjadwalan tugas dengan garpu-gabung ke utas tidak akan terpengaruh. 3. Belum mengetahui bagaimana stream paralel memetik thread dari fork-join (tidak dapat menemukan dokumentasi yang tepat di dalamnya), jadi coba gunakan pabrik threadNaming yang berbeda untuk memastikan, jika thread dalam stream paralel diambil dari customThreadFactory yang Anda berikan. 4. commonThreadPool tidak akan menggunakan customThreadFactory ini.