[java] Java 8 병렬 스트림의 사용자 정의 스레드 풀

병렬 스트림은 기본 사용 하면 프로세서가 기본적으로 하나 개의 적은 스레드가을 에 의해 반환, (그들은 또한 메인 스레드를 사용하기 때문에 병렬 스트림의 모든 프로세서를 사용하는 것이이 방법을) :ForkJoinPool.commonPoolRuntime.getRuntime().availableProcessors()

별도 또는 사용자 지정 풀이 필요한 응용 프로그램의 경우 지정된 대상 병렬 수준으로 ForkJoinPool을 구성 할 수 있습니다. 기본적으로 사용 가능한 프로세서 수와 같습니다.

또한 중첩 된 병렬 스트림 또는 여러 병렬 스트림이 동시에 시작된 경우 모두 동일한 풀을 공유 합니다. 장점 : 기본값 (사용 가능한 프로세서 수) 이상을 사용하지 마십시오. 단점 : 시작한 각 병렬 스트림에 “모든 프로세서”가 할당되지 않을 수 있습니다 (하나 이상이있는 경우). 분명히 ManagedBlocker 를 사용하여 이를 피할 수 있습니다.

병렬 스트림이 실행되는 방식을 변경하려면 다음 중 하나를 수행하십시오.

• 병렬 스트림 실행을 자신의 ForkJoinPool에 제출하십시오. yourFJP.submit(() -> stream.parallel().forEach(soSomething)).get();또는
• System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20")20 개 스레드의 대상 병렬 처리에 대해 시스템 특성을 사용하여 공통 풀의 크기를 변경할 수 있습니다 . 그러나 백 포트 패치 https://bugs.openjdk.java.net/browse/JDK-8190974 후에는 더 이상 작동하지 않습니다 .

프로세서가 8 개인 내 컴퓨터의 예입니다. 다음 프로그램을 실행하면

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

출력은 다음과 같습니다.

215216216216216216216216216315316316316316316316316415416416416416

따라서 병렬 스트림이 한 번에 8 개의 항목을 처리한다는 것을 알 수 있습니다. 즉, 8 개의 스레드를 사용합니다. 그러나 주석 처리 된 행의 주석을 해제하면 출력은 다음과 같습니다.

215215215215215215216216216216216216216216216216216216216216216216216

이번에는 병렬 스트림이 20 개의 스레드를 사용했으며 스트림의 20 개 요소가 모두 동시에 처리되었습니다.

답변

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

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

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

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

5 // common pool
23 // custom pool

3 // common pool
4 // custom pool

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