Parallel Stream Vs CompletableFuture

I am going to discuss here CompletableFuture vs Parallel Stream in Java programming language. CompletableFuture extends Future with added advantage to allow the tasks finish in an ad hoc manner, whereas, in Parallel Stream task is divided into sub-tasks and run on separate threads to be completed faster. Both CompletableFuture and Parallel Stream were added in Java 8.

I am not going to explain in details about CompletableFuture and Parallel Stream here, but you can always check the corresponding link to read details. I am going to tell you in which situations one can be used over the other because both CompletableFuture and Parallel Stream API are used to finish tasks parallelly. I am also going to show you the performance comparison between Parallel Stream and CompletableFuture.

Performance could be similar while both CompletableFuture and Parallel Stream use the same fork join common pool. The performance of Completable Future could be better in circumstances where you need to configure custom Executor with a chosen number of threads.

Parallel Stream could be your best choice for CPU intensive tasks where you want every task is doing some work, or in other words every thread is performing the same activity with different data. Parallel stream is also easier to use.

If you are looking for an asynchronous way of performing some activities, for example, you want to download from a URL while doing something else, then you can choose Completable Future API:

CompletableFuture<Object> downloadData(URL url);

Now I am going to show you with examples to compare performance on sequential, parallel and completable future.

Example – Email Notifications

Let’s say you want to send email notifications to hundred (100) customers on new updates. I am generating random email addresses in Java program. The whole source code can be downloaded later from this tutorial.

public class EmailNotifier {

	public void sendEmailNotification(final String email) {
		// send email to the customer
		System.out.println(Thread.currentThread().getName());
		try {
			Thread.sleep(1000); // for example only
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}

Sequential Stream

First I am going to check how much time it takes send email notifications to 100 customers using sequential stream.

public class SequentialStreamApp {

	public static void main(String[] args) {
		List<String> emailList = EmailUtils.getEmailList(100);

		EmailNotifier emailNotifier = new EmailNotifier();

		long start = System.nanoTime();

		emailList.stream().forEach(email -> emailNotifier.sendEmailNotification(email));

		long duration = (System.nanoTime() - start) / 1_000_000;

		System.out.printf("SequentialStreamApp: Sent %d emails in %d millis\n", emailList.size(), duration);
	}

}

The above program gives the following output:

main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
main
SequentialStreamApp: Sent 100 emails in 100947 millis

So the sequential execution will take 100 seconds to run the tasks.

Parallel Stream

Next I am going to check how much time it takes using Java’s parallel stream API.

public class ParallelStreamApp {

	public static void main(String[] args) {
		List<String> emailList = EmailUtils.getEmailList(100);

		EmailNotifier emailNotifier = new EmailNotifier();

		long start = System.nanoTime();

		emailList.parallelStream().forEach(email -> emailNotifier.sendEmailNotification(email));

		long duration = (System.nanoTime() - start) / 1_000_000;

		System.out.printf("ParallelStreamApp: Sent %d emails in %d millis\n", emailList.size(), duration);
	}

}

Running the above code will give you the following output:

main
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-7
main
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
main
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
main
ForkJoinPool.commonPool-worker-3
main
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
main
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
main
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
main
ForkJoinPool.commonPool-worker-3
main
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
main
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
main
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
main
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
main
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-5
ParallelStreamApp: Sent 100 emails in 31311 millis

The parallel stream takes 31 seconds to run the tasks. You see in the above output maximum 50 tasks were executed parallelly and there were several other tasks which executed parallelly also.

Completable Future

Using Java’s completable future now I am going to check how much time it takes.

public class CompletableFutureApp {

	public static void main(String[] args) {
		List<String> emailList = EmailUtils.getEmailList(100);

		EmailNotifier emailNotifier = new EmailNotifier();

		long start = System.nanoTime();

		emailList.stream()
				.forEach(email -> CompletableFuture.runAsync(() -> emailNotifier.sendEmailNotification(email)));

		long duration = (System.nanoTime() - start) / 1_000_000;

		System.out.printf("CompletableFutureApp: Sent %d emails in %d millis\n", emailList.size(), duration);
	}
}

Running the above code you will get the following output:

ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-5
ForkJoinPool.commonPool-worker-7
ForkJoinPool.commonPool-worker-3
CompletableFutureApp: Sent 100 emails in 34361 millis

Here you see it only uses 3 ForkJoinPool threads to execute the tasks. It took 34 seconds. Here you also notice that the main thread was not executed, which was the case for parallel stream execution.

So in this implementation parallel stream and CompletableFuture have almost the same performance, because they use the same thread pool with the number equals to Runtime.getRuntime().availableProcessors().

One of the advantages of completable future over parallel stream is that it allows you to specify your own Executor to submit the tasks. So can choose to increase or decrease the number of threads based on your application requirements.

public class CompletableFutureExecutorApp {

	public static void main(String[] args) {
		List<String> emailList = EmailUtils.getEmailList(100);

		EmailNotifier emailNotifier = new EmailNotifier();

		ExecutorService executor = Executors.newFixedThreadPool(10);

		long start = System.nanoTime();

		CompletableFuture[] futures = emailList.stream().map(email -> CompletableFuture.supplyAsync(() -> email))
				.map(future -> future.thenAcceptAsync(email -> emailNotifier.sendEmailNotification(email), executor))
				.toArray(CompletableFuture[]::new);

		CompletableFuture.allOf(futures).join();

		long duration = (System.nanoTime() - start) / 1_000_000;

		System.out.printf("CompletableFutureExecutorApp: Sent %d emails in %d millis\n", emailList.size(), duration);
	}
}

Now running the above program will give you the following output:

pool-1-thread-7
pool-1-thread-9
pool-1-thread-4
pool-1-thread-1
pool-1-thread-3
pool-1-thread-6
pool-1-thread-8
pool-1-thread-5
pool-1-thread-10
pool-1-thread-2
pool-1-thread-6
pool-1-thread-7
pool-1-thread-2
pool-1-thread-1
pool-1-thread-8
pool-1-thread-9
pool-1-thread-5
pool-1-thread-4
pool-1-thread-10
pool-1-thread-3
pool-1-thread-5
pool-1-thread-4
pool-1-thread-9
pool-1-thread-7
pool-1-thread-8
pool-1-thread-3
pool-1-thread-10
pool-1-thread-1
pool-1-thread-6
pool-1-thread-2
pool-1-thread-1
pool-1-thread-8
pool-1-thread-6
pool-1-thread-2
pool-1-thread-10
pool-1-thread-5
pool-1-thread-4
pool-1-thread-7
pool-1-thread-9
pool-1-thread-3
pool-1-thread-9
pool-1-thread-3
pool-1-thread-1
pool-1-thread-8
pool-1-thread-2
pool-1-thread-6
pool-1-thread-7
pool-1-thread-4
pool-1-thread-5
pool-1-thread-10
pool-1-thread-1
pool-1-thread-8
pool-1-thread-2
pool-1-thread-6
pool-1-thread-7
pool-1-thread-3
pool-1-thread-5
pool-1-thread-10
pool-1-thread-9
pool-1-thread-4
pool-1-thread-9
pool-1-thread-8
pool-1-thread-2
pool-1-thread-7
pool-1-thread-6
pool-1-thread-3
pool-1-thread-5
pool-1-thread-10
pool-1-thread-4
pool-1-thread-1
pool-1-thread-7
pool-1-thread-10
pool-1-thread-3
pool-1-thread-6
pool-1-thread-2
pool-1-thread-8
pool-1-thread-9
pool-1-thread-1
pool-1-thread-4
pool-1-thread-5
pool-1-thread-10
pool-1-thread-3
pool-1-thread-6
pool-1-thread-1
pool-1-thread-8
pool-1-thread-9
pool-1-thread-2
pool-1-thread-7
pool-1-thread-4
pool-1-thread-5
pool-1-thread-10
pool-1-thread-7
pool-1-thread-1
pool-1-thread-9
pool-1-thread-3
pool-1-thread-6
pool-1-thread-8
pool-1-thread-4
pool-1-thread-5
pool-1-thread-2
CompletableFutureExecutorApp: Sent 100 emails in 10138 millis

Now it took just 10 seconds to complete the execution of tasks with 10 threads.

Therefore you should consider to use CompletableFuture if your tasks are I/O intensive as it provides more control over the number of threads. But for CPU-intensive tasks go for the Parallel stream as there is no point of having more threads than the number of processors.

Source Code

Download