public static < T > void throttledExecution ( Stream < T > stream , Consumer < T > consumer , ConcurrencyConfiguration concurrencyConfiguration ) { if ( concurrencyConfiguration . isParallelized ( ) ) { int maxThreads = concurrencyConfiguration . getMaxNumberOfThreadsPerTask ( ) ; int maxTasks = 2 * maxThreads ; ExecutorService executorService = Executors . newFixedThreadPool ( maxThreads ) ; ThrottledExecutor executor = new ThrottledExecutor ( executorService , maxTasks ) ; stream . sequential ( ) . forEach ( i -> { executor . execute ( ( ) -> { consumer . accept ( i ) ; } ) ; } ) ; executorService . shutdown ( ) ; try { executorService . awaitTermination ( Integer . MAX_VALUE , TimeUnit . SECONDS ) ; } catch ( InterruptedException ex ) { throw new RuntimeException ( ex ) ; } } else { Runnable runnable = ( ) -> stream . forEach ( consumer ) ; runnable . run ( ) ; } }
Takes the items of the stream in a throttled way and provides them to the consumer . It uses as many threads as the available processors and it does not start more tasks than 2 times the previous number .