For loop including if to parallelStream() expression

Question

Is there a way to parallelize this piece of code:

HashMap<String, Car> cars;
List<Car> snapshotCars = new ArrayList<>();
...
for (final Car car : cars.values()) {
    if (car.isTimeInTimeline(curTime)) {
        car.updateCalculatedPosition(curTime);
        snapshotCars.add(car);
    }
}

Update: This is what I tried before asking for assistance:

snapshotCars.addAll(cars.values().parallelStream()
                                 .filter(c -> c.isTimeInTimeline(curTime))
                                 .collect(Collectors.toList()));

How could I integrate this line? -> car.updateCalculatedPosition(curTime);

Have you tried converting to the stream API? What problems did you encounter? — khelwood, Jun 06 '16 at 11:22
Sure is. Now you should try. Don't expect us to do your work for you. — GhostCat, Jun 06 '16 at 11:23
@RealSkeptic I want you to tell me how to write it. Small semantic difference I guess. — Simon S., Jun 06 '16 at 12:12

score 1 · Answer 1 · edited May 23 '17 at 10:29

Well, assuming that updateCalculatedPosition does not affect state outside of the Car object on which it runs, it may be safe enough to use peek for this:

List<Car> snapshotCars = cars.values()
    .parallelStream()
    .filter(c -> c.isTimeInTimeline(curTime))
    .peek(c -> c.updateCalculatedPosition(curTime))
    .collect(Collectors.toCollection(ArrayList::new));

I say this is "safe enough" because the collect dictates which elements will be peeked by peek, and these will necessarily be all the items that passed the filter. However, read this answer for the reason why peek should generally be avoided for "significant" operations.

Your peek-free alternative is to first, filter and collect, and then update using the finished collection:

List<Car> snapshotCars = cars.values()
    .parallelStream()
    .filter(c -> c.isTimeInTimeline(curTime))
    .collect(Collectors.toCollection(ArrayList::new));
snapShotCars.parallelStream()
    .forEach(c -> c.updateCalculatedPosition(curTime));

This is safer from an API point of view, but less parallel - you only start updating the positions after you have finished filtering and collecting.

Unless the OP says explicitly that an `ArrayList` is needed, I would simply use `Collectors.toList()`. — Holger, Jun 07 '16 at 10:06

score 0 · Answer 2 · answered Jun 06 '16 at 12:41

If you want parallelized access to a List you might want to use Collections.synchonizedList to get a thread-safe list:

List<Car> snapshotCars = Collections.synchronizedList(new ArrayList<>());

Then you can use the stream API like so:

cars.values()
    .parallelStream()
    .filter(car -> car.isTimeInTimeline(curTime))
    .forEach(car -> {
        car.updateCalculatedPosition(curTime);
        snapshotCars.add(car);
    });

score 0 · Answer 3 · edited May 23 '17 at 12:26

In addition to RealSkeptic’s answer, you can alternatively use your own collector:

List<Car> snapshotCars = cars.values().parallelStream()
    .filter(c -> c.isTimeInTimeline(curTime))
    .collect(ArrayList::new,
             (l,c) -> { c.updateCalculatedPosition(curTime); l.add(c); },
             List::addAll);

Note that .collect(Collectors.toList()) is equivalent (though not necessarily identical) to .collect(Collectors.toCollection(ArrayList::new)) which is equivalent to .collect(ArrayList::new, List::add, List::addAll).

So our custom collector does a similar operation, but replaces the accumulator with a function, which also performs the desired additional operation.

For loop including if to parallelStream() expression

3 Answers3