使用并行流返回提供的最快值
2022-09-03 14:16:36
我有一组供应商,它们都支持相同的结果,但速度不同(和不同)。
我想要一种优雅的方式来同时启动供应商,一旦其中一个产生了价值,就将其返回(丢弃其他结果)。
我尝试过使用并行流和为此,但它似乎总是阻塞,直到产生所有结果。Stream.findAny()
下面是一个单元测试,演示了我的问题:
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
}
}
测试的结果是最后一个断言失败,因为整个测试大约需要 10 秒才能完成。
我在这里做错了什么?