Java stream() vs parallel stream()

Just a point of view on when to use both of these approaches.

Since Java 8, everybody knows and uses a stream(). But what will happen if we make a transformation process in parallel, with threads per CPU core. Parallel Stream is a great idea, but we need to consider the possible pitfalls:

1. Operations like forEach() don’t guarantee encounter order in parallel streams.
2. Parallelism works well when we iterate per element, when the source is easily splittable: ArrayList, IntStream.range(), LongStream.range().
3. We can easily get a race condition if using a non-thread-safe ArrayList to modify shared state. It is not synchronized, so multiple threads are trying to add elements simultaneously.

// List<Integer> result = Collections.synchronizedList(new ArrayList<>());
List<Integer> result = new ArrayList<>();
IntStream.range(0, 100).parallel().forEach(result::add); // race condition

// use thread-safe collection instead of:
List<Integer> safeResult = IntStream.range(0, 100) // primitive type
    .parallel()
    .boxed()
    .collect(Collectors.toList()); 

4. Use custom pools for CPU-intensive operations that need isolation.

// Use custom pool for big parallel work operation
public List<Integer> processInParallel(List<Integer> input) {
    // define threads count
    ForkJoinPool customPool = new ForkJoinPool(
                              Runtime.getRuntime().availableProcessors());

    List<Integer> result = customPool.submit(() -> 
        IntStream.range(0, 100).parallel()
        .map(i -> i * 2)
        .boxed()
        .collect(Collectors.toList())
    ).get();

    return customPool.submit(() -> 
        input.parallelStream()
            .map(this::expensiveOperation)
            .collect(Collectors.toList())
    ).join();
}

5. Use CompletableFuture for more control over parallelism

List<CompletableFuture<Integer>> futures = IntStream.range(0, 100)
    .mapToObj(i-> CompletableFuture.supplyAsync(()-> process(i),customPool))
    .collect(Collectors.toList());

List<Integer> results = futures.stream()
    .map(CompletableFuture::join)
    .collect(Collectors.toList());

6. Use the right collection.

a. toConcurrentMap() — thread-safe map collection, when you need to collect stream elements into a Map in parallel. This map contains 3 arguments: keyMapper, valueMapper, and mergeFunction.

List<String> words = Arrays.asList("A", "B", "A", "C", "D", "E");
Map<String, Integer> wordCounts = words.parallelStream()
    .collect(Collectors.toConcurrentMap(
        word -> word,
        word -> 1,
        Integer::sum
    ));

b. groupingByConcurrent() — Parallel Grouping when you need to group elements by a classifier function in parallel.

ConcurrentMap<String, Long> countByCity = people.parallelStream()
    .collect(Collectors.groupingByConcurrent(
        Person::getCity,
        Collectors.counting()
    ));

// in ordinary stream we use such mapping:
Map<Character, Long> caseInsensitiveCount = input.chars()
    .mapToObj(c -> (char) Character.toLowerCase(c))
    .collect(Collectors.groupingBy(
        Function.identity(),
        Collectors.counting()
    ));

7. Preserve the order anyway:

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
numbers.parallelStream().map(n -> n * 2)
       .forEachOrdered(System.out::println);

8. Difference between stream().parallel() and parallelStream()

// stream().parallel() just convert stream to parallel

List<String> list = Arrays.asList("a", "b", "c");
Stream<String> parallelStream = list.stream().parallel();

// parallelStream() implemented just right over the object of collection.

List<String> list = Arrays.asList("a", "b", "c");
Stream<String> parallelStream = list.parallelStream();

9. Write a simple time elapsed check for the test, because parallelStream() will not always work faster due to the expenses of thread control.

long start = System.nanoTime();
List<Integer> list= Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
int sum = list.parallelStream().mapToInt(Integer::intValue).sum(); //primitive
long duration = System.nanoTime() - start;
System.out.println("Parallel took: " + duration + " ns");

This approach reminds me of the RxJS and RxJava libraries, similar to working 1 thread per CPU core. So, use parallel streams for CPU-bound tasks.

Get Aleksei Jegorov’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Good candidates for parallel processing:

• Big datasets
• Stateless operations
• Independent operations
• CPU-intensive operations

Java map vs flatMap vs peek

In most cases, regular work with stream() is based on simple map transformations. Parallel stream is good for special cases only.

1. map() for 1:1 transformations — returns new stream type.
2. flatMap() for 1:many transform scenarios and flattening.
3. map() keep the order, but flatMap() does not even try.
4. flatMap use case: handling collections of collections.
5. peek() use for observation only — don’t use for mutations.

List<String> names = Arrays.asList("A", null, "B", "C", null);
// remove nulls to avoid innormal output
List<String> result = names.stream()
    .filter(item -> !item.isEmpty())
    .map(String::toUpperCase)
    //.filter(Objects::nonNull)
    .collect(Collectors.toList());

flatMap returns Stream<R> where each input element maps to a stream.

List<List<String>> nestedList = Arrays.asList(
    Arrays.asList("a", "b"), Arrays.asList("c", "d")
);
List<String> flatList = nestedList.stream()
    .flatMap(List::stream).collect(Collectors.toList());
// ["a", "b", "c", "d"]

Performance Considerations between operators

// flatMap creates many small streams, so it is not efficiently
List<String> result = list.stream()
    .flatMap(item -> Arrays.stream(item.split(",")))
    .collect(Collectors.toList());

// more efficient for large datasets
List<String> result = new ArrayList<>();
list.forEach(item -> {
    String[] parts = item.split(","); Collections.addAll(result, parts);
});

Object Flattening

class User {
    private List<String> roles;
    public List<String> get() { return roles; }
}
List<User> users = new ArrayList<>();
Set<String> allRoles = users.stream()
    .map(User::getRoles)             // Get all unique roles from all users
    .flatMap(List::stream)           // roles list to stream
    .collect(Collectors.toSet());

The Peek() operator should not be used for production mode; it was developed mostly for the debugging process:

List<String> result = dataList.stream()
    .peek(item -> log.debug("Processing: {}", item))
    .map(complexLongTransformation)
    .peek(result -> log.debug("Transformed: {}", result))
    .collect(Collectors.toList());

// Or use a wrapper method:
private <T> T logAndTransform(T item, Function<T, T> transformer) {
    log.debug("Processing: {}", item);
    T result = transformer.apply(item);
    log.debug("Result: {}", result);
    return result;
}
List<String> result = dataList.stream()
    .map(item -> logAndTransform(item, this::complexTransformation))
    .collect(Collectors.toList());

In case of remembrance of RxJS on front-end: it has a special method tap(), which works in the same way as peek() for debugging and logging — just pass the Observable stream without side effects.

import { fromFetch } from 'rxjs/fetch';
import { switchMap, catchError, tap } from 'rxjs/operators';
function getUserData(userId: number) {
  return fromFetch(`/api/users/${userId}`)
    .pipe(
      tap(response => {
        if (!response.ok) {
          throw new Error('Network request failure');
        }
      }),
      switchMap(response => response.json()),
      tap(userData => {
        loadingService.setLoading(false);
        cacheService.set(`usr_${userId}`, userData);
      }),
      catchError(error => {
        errorService.handle(error);
        loadingService.setLoading(false);
        throw error;
      })
    );
}

Conclusion

Parallel stream in Java — this is a special type of stream that allows you to perform operations on collection elements in parallel, that is, simultaneously in several streams.

• Increase performance and speed up the processing of data.
• Distribute the workload across multiple threads, which reduces processing time. Automatically splits the collection into several parts and processes each part in a separate thread. The results are then combined into one overall result.
• Use all available processor cores to perform operations.

It requires careful consideration of thread safety, performance implications, and appropriate use cases.