# Using Java 8 parallel streams

I'm trying to get more familiar with the new Java 8 features, so I am rewriting one of my earlier projects. It includes a class that keeps track of the best entry seen so far:

import java.util.List;

public class Maxinator<T> {

private final QualityFunction<T> qualityFunction;
private T best;
private double bestQuality;

public Maxinator(QualityFunction<T> qualityFunction) {
this.qualityFunction = qualityFunction;
reset();
}

public void reset() {
best = null;
bestQuality = Double.NEGATIVE_INFINITY;
}

public T getBest() {
return best;
}

public void updateBest(List<T> population) {
population.parallelStream()
.forEach(i -> {
double quality = qualityFunction.computeQuality(i);
if (quality > bestQuality) {
best = i;
bestQuality = quality;
}
});
}
}


Where QualityFunction is simply the following interface:

public interface QualityFunction<T> {
public abstract double computeQuality(T individual);
}


My questions:

• Are there any concurrency issues with the code in the forEach modifying best and bestQuality? Or are these automatically taken care of by the parallel stream processing?
• Is there a more idiomatic way to write this, using the new APIs (collect, reduce, etc.)?

Note: the quality function could be very expensive, so I want to make sure that it is only called once per element.

-

Your code is not thread-safe. Each of the threads will, in parallel, be accessing both the best, and the bestQuality variables.

Your Lambda is, in essence, modifying external data from the stream, and this is an anti-pattern for streams. It has side-effects.

You should change your code to use the collect mechanism. There are a few ways to do it, but, you should look at this example for guidance: Reduction

Some Notes about my following suggestion:

• I converted the method to a static method, and created an inner accumulator class.
• you should possibly refine this answer to suit your needs more in the class structure.

Putting these observations together, I would suggest something like:

public class Maxinator {

@FunctionalInterface
public interface QualityFunction<V> {
double computeQuality(V value);
}

private Maxinator() {
//        no public instances
}

private static final class AccumulateResult<T> {
T bestItem = null;
double bestScore = Double.MIN_VALUE;

public T getBestItem() {
return bestItem;
}

public void accept(QualityFunction<T> function, T item) {
double score = function.computeQuality(item);
if (score > bestScore) {
bestScore = score;
bestItem = item;
}
}

public AccumulateResult<T> combine(AccumulateResult<T> r) {
if (r.bestScore > bestScore) {
bestScore = r.bestScore;
bestItem = r.bestItem;
}
return this;
}

}

public static <T> T getBest(final QualityFunction<T> qualityFunction, final List<T> population) {
return population.parallelStream().collect(Collector.of(
AccumulateResult<T>::new,
(a,t) -> a.accept(qualityFunction, t),
(a, b) -> a.combine(b))
).getBestItem();
}
}

-
Updated with a code example using an accumulator. –  rolfl Aug 18 '14 at 19:09
I knew there had to be a more natural way to fit this into the streams framework, and this is it. Thank you! I do have two questions: 1. Why did you go for Double.MIN_VALUE, instead of Double.NEGATIVE_INFINITY as starting value? 2. If I really want my class to keep track of the best so far (over multiple calls of updateBest), I assume I can use this same idea, but initialize the accumulator with the current bestItem and bestScore? –  Mangara Aug 18 '14 at 19:20
Double.NEGATIVE_INFINITY would be better, but really, I should have a boolean first flag, and go from there. As for the keeping-the-old value - you should realize that when streaming the values, there will be many 'Result' instances, and they will all be merged as the parallel threads complete. –  rolfl Aug 18 '14 at 20:59
Will this result in a parallel computation? The docs suggest that this only happens if the Collector defines the Collector.Characteristics.CONCURRENT and Collector.Characteristics.UNORDERED characteristics. –  Mangara Aug 18 '14 at 21:16
The Collector does not need to be concurrent for the stream to be parallel. The collector has an efficient 'combine' method, which makes the non-concurrent collector efficient still. Instead of collecting all the values in one concurrent collector, the parallel stream collects in to multiple collectors, and then safely merges the results. –  rolfl Aug 19 '14 at 0:31

http://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#max-java.util.Comparator-

Returns the maximum element of this stream according to the provided Comparator. This is a special case of a reduction.

Furthermore, there's a handy static method on Comparator to lift a regular function to a comparator:

http://docs.oracle.com/javase/8/docs/api/java/util/Comparator.html#comparing-java.util.function.Function-

Accepts a function that extracts a Comparable sort key from a type T, and returns a Comparator that compares by that sort key.

e.g.,

public static <T> T getBest(final Function<T, Double> qualityFunction, final List<T> population) {
return population.parallelStream()
.max(Comparator.comparing(qualityFunction))
.get(); // unwrap Optional<T>
}


The java.util.function package has some nice stuff in it (though not quite as rich as, say, Scala or clojure), but you have to hunt through the javadocs for a lot of it.

-
This is a really nice solution! My only concern is the number of times qualityFunction is evaluated here. In my test, it was evaluated roughly twice for each object. As some quality functions in my project are really expensive, this would effectively double the running time. –  Mangara Aug 18 '14 at 20:45
Yep, unfortunately the implementation of max and Comparator.comparing are naive, and don't do any caching, so you'd be better off making your own Collector which caches the result. The docs on MutableReduction describe this tradeoff. –  blendmaster Aug 18 '14 at 21:02
Note that there is also Collectors.maxBy(Comparator<T>), but unfortunately has the same problem as Stream#max in that its implementation defers to Collectors#reducing (i.e., Stream#reduce) and therefore can't generically cache the Comparator.comparing(qualityFunction). –  blendmaster Aug 18 '14 at 21:08