병렬 스트림에서의 동시 평가 (fixedThreadPool과 같은)의 수를 제한하는 최선의 방법은 무엇입니까?

Question

람다 표현식은 메모리와 같은 특정 양의 리소스를 소비한다고 가정하고 동시 실행 수 (예 : 람다가 로컬 메모리의 100MB를 일시적으로 소비하고이를 1GB로 제한하려는 경우 10 개의 동시 평가를 허용하지 않습니다).

동시 실행의 수를 제한 할 수있는 가장 좋은 방법은 무엇입니까,

IntStream.range(0, numberOfJobs).parallel().foreach(i -> { /*...*/ }); 

, 예를 들어 말을?

참고 : 확실한 옵션은

double jobsPerThread = (double)numberOfJobs/numberOfThreads; 
    IntStream.range(0, numberOfThreads).parallel().forEach(threadIndex -> 
     IntStream.range((int)(threadIndex * jobsPerThread), (int)((threadIndex+1) * jobsPerThread)).sequential().forEach(i -> { /*...*/ })); 

이 유일한 방법입니다 같은 중첩을 수행하는 것입니다? Tt는 이 아니며 우아한입니다. 사실은 내가 쉽게 할 수있는 CompletableFuture 유틸리티 방법을 사용하여, 사용 사례에 따라,

IntStream.range(0, numberOfJobs).parallel(numberOfThreads).foreach(i -> { /*...*/ }); 

Answer 1

을 가지고 싶다 :

import static java.util.concurrent.CompletableFuture.runAsync; 

ExecutorService executor = Executors.newFixedThreadPool(10); //max 10 threads 
for (int i = 0; i < numberOfJobs; i++) { 
    runAsync(() -> /* do something with i */, executor); 
} 

//or with a stream: 
IntStream.range(0, numberOfJobs) 
     .forEach(i -> runAsync(() -> /* do something with i */, executor)); 

코드와 가장 큰 차이점은 그 병렬 forEach만이 반환 마지막 작업이 끝난 후 runAsync이 모든 작업이 제출되면 즉시 반환됩니다. 필요한 경우 해당 동작을 변경하는 다양한 방법이 있습니다.

Answer 2

Stream은 병렬 작업에 ForkJoinPool을 사용합니다. 기본적으로 그들은 ForkJoinPool.commonPool()을 사용하고 있으며, 이후에는 동시성을 변경할 수 없습니다. 그러나 자신 만의 ForkJoinPool 인스턴스를 사용할 수 있습니다. 자신의 ForkJoinPool 컨텍스트 내에서 스트림 코드를 실행하면이 컨텍스트 풀이 스트림 작업에 사용됩니다.

import java.util.HashSet; 
import java.util.concurrent.ExecutionException; 
import java.util.concurrent.ForkJoinPool; 
import java.util.stream.IntStream; 

public class InterfaceStaticMethod { 
    public static void main(String[] arg) throws Exception { 
     Runnable parallelCode=() -> { 
     HashSet<String> allThreads=new HashSet<>(); 
     IntStream.range(0, 1_000_000).parallel().filter(i->{ 
      allThreads.add(Thread.currentThread().getName()); return false;} 
     ).min(); 
     System.out.println("executed by "+allThreads); 
     }; 
     System.out.println("default behavior: "); 
     parallelCode.run(); 
     System.out.println("specialized pool:"); 
     ForkJoinPool pool=new ForkJoinPool(2); 
     pool.submit(parallelCode).get(); 
    } 
}