Parallel Stream adding items to same array

Question

I have an array of Objects which I stream to use a function on each object to modify it and add it to a completely new array of Objects.

By using .parallel() the execution time speeds up 2 times, but I'm getting some NullPointerExceptions when looping over the new array.

I tried to debug it a few times without any success. It seems that this problem only occurs during the runtime. I tried implementing a synchronized function for adding the new objects to the list, but sadly it didn't work as well.

Can anyone give me a suggestion how to get this working? Thanks in advance!

Here is the code snippet:

private static final Object sync = new Object();
private ArrayList<Object> newList = new ArrayList<Object>();      

private void addNewObject(Object newObject) {
        synchronized (sync) {
            newList.add(newObject);
        }
    }

private Object mutateObject(Object oldObject) {
    // Do something with the object here
    return mutatedObject;
}

public ArrayList<Object> createNewList(ArrayList<Object> oldList) {
    oldList.stream().parallel().forEach(object -> addNewObject(mutateObject(object)));
    return newList;
}

Why do you add an parameter to the createNewList function without using it? — fireandfuel
– fireandfuel, Commented Apr 7, 2018 at 14:35
Consider using the built-in concurrent collections so that you don't have to write your own concurrency code. Also, use interfaces like List<Object> for your public interface instead of implementations like ArrayList — Novaterata
– Novaterata, Commented Apr 7, 2018 at 16:53

Roy · Accepted Answer · 2018-04-07 16:42:33Z

5

Consider refactoring createNewList method to something like this:

public List<Object> createNewList(List<Object> oldList) {
    return oldList.parallelStream()
            .map(this::mutateObject)
            .collect(Collectors.toList());
}

If you look up into the java.util.stream.Stream JavaDoc you can see that there is a few terminal operations, like forEach, collect, toArray, etc. In your question it's better to use the collect terminal operation because it handles synchronization internally, avoiding race condition situations, without any further implementation of synchronization.

Mapping objects with .map() (an intermediate operation of Stream) and collecting result to a final list in parallel solves this problem.

The addNewObject function and sync object should be removed, because it's not used anymore.

edited Apr 7, 2018 at 16:42

Roy

8,1295 gold badges29 silver badges49 bronze badges

answered Apr 7, 2018 at 14:30

Szymon Stepniak

42.6k10 gold badges117 silver badges140 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

fireandfuel Over a year ago

You can replace object -> mutateObject(object) with this::mutateObject.

Robert Over a year ago

Thank you for the fast response! I'll try it. What can I do if I would like to add a object multiple times to the new list?

Szymon Stepniak Over a year ago

@Robert You can always return a list from .map() (e.g. .map(object -> Arrays.asList(mutateObject(object), mutateObject(object), ....)) operation and do.flatMap(Collection::stream) on the stream to transform list of lists to a single list.

Holger Over a year ago

@SzymonStepniak there’s no sense in mapping to a List, just to do .flatMap(Collection::stream) afterwards. Just do .flatMap(object -> Stream.of(mutateObject(object), mutateObject(object), ....)) in the first place…

Collectives™ on Stack Overflow

Parallel Stream adding items to same array

1 Answer 1

4 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

4 Comments

Your Answer

Sign up or log in

Post as a guest

Related