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

【问题标题】：First Object in Set<Future<Object>> that satisfies a predicateSet<Future<Object>> 中满足谓词的第一个对象
【发布时间】：2015-08-25 17:09:33
【问题描述】：

抽象概念

我想从一组 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 来实现它，但我都觉得这个问题非常不习惯，我最终得到了很多丑陋的代码。

【问题讨论】：

标签： java functional-programming java-8 future rx-java

【解决方案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);

【讨论】：

【解决方案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));
}

【讨论】：