Set<Future<Object>> 中满足谓词的第一个对象

Question

抽象思想

我想从一组 Futures 中获取满足给定谓词的第一个值。

如果找到满意的值，则应取消所有其他期货。如果在所有 Futures 返回后没有找到值，则应终止执行（通过返回默认值或抛出异常）。

具体例子

public boolean isThereVacantHotelRooms(Set<URL> hotelApiUrls) {
    // returns true if any provided server responds with a number larger than 0
}

我正在寻找一种在 Java 8 中实现上述内容的漂亮方法（外部库很好）。我已经尝试用 CompletableFuture 和 RxJava 来实现它，但我都觉得这个问题很不合常理，最后我得到了很多丑陋的代码。

Answer 1

我认为，您的案例可以通过合并、过滤和获取的组合来完成：

List<Observable<HotelInfo>> hotels = new ArrayList<>();
for (URL u : urls) {
    Observable<HotelInfo> hotelInfo = networkAPI.askHotel(u);
    hotels.add(hotelInfo);
}
Observable.merge(hotels)
.filter(h -> h.vacancy > 0)
.take(1)
.subscribe(h -> System.out.println("Winner: " + h), Throwable::printStackTrace);

Answer 2

由于您尝试过其他方案，这里提供bayou方案进行比较

// throw exception for failure
public void checkVacantHotelRooms(Set<URL> hotelApiUrls) throws Exception
{
    checkVacantHotelRoomsAsync(hotelApiUrls)  // Async<Void>
        .timeout(Duration.ofSeconds(10))      // cancel on 10s
        .sync();    //  async -> sync
}

public Async<Void> checkVacantHotelRoomsAsync(Set<URL> hotelApiUrls)
{
    Stream<Async<Void>> resultStream = hotelApiUrls.stream()    // Stream<URL>
        .map(this::getResponseBodyAsync)                        // Stream<Async<String>>
        .map(asyncBody->asyncBody.map(this::checkResponse));    // Stream<Async<Void>

    return AsyncBundle.anyOf(resultStream); 
    // succeeds if one result succeeds; others will be cancelled
}

Void checkResponse(String responseBody) throws Exception
{
    if(responseBody.contains("something"))
        return (Void)null;
    throw new Exception("none in this hotel");
}

-----

HttpClient httpClient = new HttpClient();
int maxBodyLength = 1000;

Async<String> getResponseBodyAsync(URL url)
{
    Async<HttpResponse> asyncResponse = httpClient.doGet(url.toExternalForm());
    return asyncResponse.then(resp->resp.bodyString(maxBodyLength));
}

Set<Future<Object>> 中满足谓词的第一个对象

First Object in Set<Future<Object>> that satisfies a predicate

java

functional-programming

future

java-8

rx-java