build a collection using java8 streams

Question

I have written this code to build a collection of objects concurrently using an Executor.newFixedThreadPool and a ConcurrentLinkedQueue. However, I find this whole thread pool and task submission to it verbose. I wish I could write it more concisely using streams.

private ConcurrentLinkedQueue<ApiRunner> getRunners(final ApiRunnerBuilder builder,
        final int testSize) throws InterruptedException {

    final ExecutorService executor = Executors.newFixedThreadPool(NUM_OF_CORES);
    ConcurrentLinkedQueue<ApiRunner> runners = new ConcurrentLinkedQueue<ApiRunner>();
    for (int i = 0; i < testSize; i++) {
        executor.execute(() -> {
            runners.add(builder.build());
        });
    }
    executor.shutdown();
    executor.awaitTermination(300, TimeUnit.SECONDS);
    return runners;
}

I'm thinking maybe it could be reduced to something like this (new to java and new to streams):

private ConcurrentLinkedQueue<ApiRunner> getRunners(final ApiRunnerBuilder builder,
        final int testSize) throws InterruptedException {

    ConcurrentLinkedQueue<ApiRunner> runners = new ConcurrentLinkedQueue<ApiRunner>();
    ConcurrentLinkedQueue<ApiRunner> runners = range(testSize).parallelStream(()-> {
            runners.add(builder.build());
        });
    }
    return runners;
}

Answer 1

Since you said, you have lots of I/O involved, I don’t recommend using the stream API. The stream API is optimized for CPU bound tasks as it adjusts the number of threads according to the number of CPU cores, but for I/O bound tasks, an even higher number of threads might be preferable as threads blocked waiting for the completion of an I/O operation do not consume CPU resources.

Note that the old API also plays nice with the new lambda expressions:

ExecutorService executor = Executors.newFixedThreadPool(DESIRED_CONCURRENCY);
ConcurrentLinkedQueue<ApiRunner> runners = new ConcurrentLinkedQueue<ApiRunner>();
executor.invokeAll(Collections.nCopies(testSize, ()->runners.add(builder.build())));
executor.shutdown();
return runners;

In real applications you might keep the executor longer than for this task. The method invokeAll already waits for the completion of all tasks, so neither shutdown nor waiting for the termination is necessary here. The shutdown invocation is only added in this example for cleaning up as in this case you’ve created the executor right inside the method.