我有一组供应商,它们都提供相同的结果,但速度不同(且不同)。
我想要一种优雅的方式来同时启动供应商,一旦其中一个产生了值(value),就返回它(丢弃其他结果)。
我已经尝试为此使用并行流和 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 秒才能完成。
我做错了什么?
最佳答案
在这种情况下,您最好使用 Callable
而不是 Supplier
(相同的功能签名)并使用自 Java 5 以来就存在的良好并发 API:
Set<Callable<String>> suppliers=new HashSet<>();
suppliers.add(() -> "fast"); // This supplier returns immediately
suppliers.add(() -> {
Thread.sleep(10_000);
return "slow";
}
);
ExecutorService es=Executors.newCachedThreadPool();
try {
String result = es.invokeAny(suppliers);
System.out.println(result);
} catch (InterruptedException|ExecutionException ex) {
Logger.getLogger(MyClass.class.getName()).log(Level.SEVERE, null, ex);
}
es.shutdown();
请注意,整个“全部运行并最快返回”是如何变成单个方法调用的……
它还有取消/中断所有未决操作的好处,只要有一个结果可用,所以缓慢的操作实际上不会在这里等待整整十秒(好吧,在大多数情况下,因为时间不是确定性的)。
关于java - 使用并行流返回最快的提供值,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/32968557/