I have a set of suppliers that all suppply the same result but with different (and varying) speed.
I want an elegant way to start off the suppliers at the same time and as soon as one of them has produced a value, return it (discarding the other results).
I've tried using parallel streams and the Stream.findAny()
for this but it always seems to block until all results have been produced.
Here's a unit test demonstrating my problem:
import org.junit.Test;
import java.util.Collections;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;
import java.util.stream.Stream;
import static org.junit.Assert.*;
public class RaceTest {
@Test
public void testRace() {
// Set up suppliers
Set<Supplier<String>> suppliers = Collections.newSetFromMap(new ConcurrentHashMap<>());
suppliers.add(() -> "fast"); // This supplier returns immediately
suppliers.add(() -> {
try {
Thread.sleep(10_000);
return "slow";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}); // This supplier takes 10 seconds to produce a value
Stream<Supplier<String>> stream = suppliers.parallelStream();
assertTrue(stream.isParallel()); // Stream can work in parallel
long start = System.currentTimeMillis();
Optional<String> winner = stream
.map(Supplier::get)
.findAny();
long duration = System.currentTimeMillis() - start;
assertTrue(winner.isPresent()); // Some value was produced
assertEquals("fast", winner.get()); // The value is "fast"
assertTrue(duration < 9_000); // The whole process took less than 9 seconds
}
}
The result of the test is that the last assertion fails as the whole test takes about 10 seconds to complete.
What am I doing wrong here?
findAny
is nondeterministic so it's just luck.Concurrent…
collection here. Any collection will do, as long as you don’t modify it during the operation. In fact, due to the way it works internally, an ordinary collection performs better at this task, even when it comes to parallel execution. But as already said by others, theStream
API is not the right tool here.