超时不起作用的可满足未来

Question

我是新的可折叠的未来。我正在尝试为元素列表(参数)调用并行方法，然后组合结果以创建最终响应。我还在尝试设置 50 毫秒的超时，以便如果调用在 50 毫秒内未返回，我将返回默认值。

到目前为止，我已经尝试过这个：

{
List<ItemGroup> result = Collections.synchronizedList(Lists.newArrayList());
try {
List<CompletableFuture> completableFutures = response.getItemGroupList().stream()
.map(inPutItemGroup -> 
CompletableFuture.runAsync(() -> {
final ItemGroup itemGroup = getUpdatedItemGroup(inPutItemGroup);               //call which I am tryin to make parallel
// this is thread safe
if (null != itemGroup) {
result.add(itemGroup); //output of the call
}
}, executorService).acceptEither(timeoutAfter(50, TimeUnit.MILLISECONDS),inPutItemGroup))  //this line throws error     
.collect(Collectors.toList());
// this will wait till all threads are completed
CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()]))
.join();
} catch (final Throwable t) {
final String errorMsg = String.format("Exception occurred while rexecuting parallel call");
log.error(errorMsg, e);
result = response.getItemGroupList(); //default value - return the input value if error
}
Response finalResponse = Response.builder()
.itemGroupList(result)
.build();
}
private <T> CompletableFuture<T> timeoutAfter(final long timeout, final TimeUnit unit) {
CompletableFuture<T> result = new CompletableFuture<T>();
//Threadpool with 1 thread for scheduling a future that completes after a timeout
ScheduledExecutorService delayer = Executors.newScheduledThreadPool(1);
String message = String.format("Process timed out after %s %s", timeout, unit.name().toLowerCase());
delayer.schedule(() -> result.completeExceptionally(new TimeoutException(message)), timeout, unit);
return result;
}

但是我不断收到错误说：

error: incompatible types: ItemGroup cannot be converted to Consumer<? super Void>
[javac]                             itemGroup))
incompatible types: inference variable T has incompatible bounds
[javac]                     .collect(Collectors.toList());
[javac]                             ^
[javac]     equality constraints: CompletableFuture
[javac]     lower bounds: Object
[javac]   where T is a type-variable:

有人可以告诉我我在这里做错了什么吗？如果我走错了方向，请纠正我。

谢谢。

Answer 1

而不是

acceptEither(timeoutAfter(50, TimeUnit.MILLISECONDS), inPutItemGroup))

你需要

applyToEither(timeoutAfter(50, TimeUnit.MILLISECONDS), x -> inPutItemGroup)

以编译代码。"accept"是使用值而不返回新值的操作，"apply"是生成新值的操作。

但是，仍然存在逻辑错误。timeoutAfter返回的未来将异常完成，因此依赖阶段也将异常完成，而不计算函数，因此此链接方法不适合将异常替换为默认值。

更糟糕的是，修复此问题将创建一个由任一源未来完成的新未来，但这不会影响在其中一个源期货中执行的result.add(itemGroup)操作。在代码中，生成的将来仅用于等待完成，而不用于评估结果。因此，当您的超时过后，您将停止等待完成，而可能仍有后台线程修改列表。

正确的逻辑是将获取值的步骤(可能会在超时时被默认值取代)与将结果(提取的值或默认值)添加到结果列表的步骤分开。然后，您可以等待所有add操作完成。超时时，可能仍有正在进行的getUpdatedItemGroup评估(无法停止其执行)，但其结果将被忽略，因此不会影响结果列表。

还值得指出的是，为每个列表元素创建一个新ScheduledExecutorService(使用后不会关闭，使事情变得更糟)不是正确的方法。

// result must be effectively final
List<ItemGroup> result = Collections.synchronizedList(new ArrayList<>());
List<ItemGroup> endResult = result;
ScheduledExecutorService delayer = Executors.newScheduledThreadPool(1);
try {
CompletableFuture<?>[] completableFutures = response.getItemGroupList().stream()
.map(inPutItemGroup ->
timeoutAfter(delayer, 50, TimeUnit.MILLISECONDS,
CompletableFuture.supplyAsync(
() -> getUpdatedItemGroup(inPutItemGroup), executorService),
inPutItemGroup)
.thenAccept(itemGroup -> {
// this is thread safe, but questionable,
// e.g. the result list order is not maintained
if(null != itemGroup) result.add(itemGroup);
})
)
.toArray(CompletableFuture<?>[]::new);
// this will wait till all threads are completed
CompletableFuture.allOf(completableFutures).join();
} catch(final Throwable t) {
String errorMsg = String.format("Exception occurred while executing parallel call");
log.error(errorMsg, e);
endResult = response.getItemGroupList();
}
finally {
delayer.shutdown();
}
Response finalResponse = Response.builder()
.itemGroupList(endResult)
.build();

private <T> CompletableFuture<T> timeoutAfter(ScheduledExecutorService es,
long timeout, TimeUnit unit, CompletableFuture<T> f, T value) {
es.schedule(() -> f.complete(value), timeout, unit);
return f;
}

在这里，supplyAsync生成一个CompletableFuture，该将提供getUpdatedItemGroup评估的结果。timeoutAfter调用将在超时后使用默认值安排完成，而不会创建新的未来，然后，通过thenAccept链接的依赖操作会将结果值添加到result列表中。

请注意，synchronizedList允许从多个线程添加元素，但从多个线程添加将导致不可预测的顺序，与源列表的顺序无关。

Answer 2

acceptEither的签名如下所示：

public CompletableFuture<Void> acceptEither(
CompletionStage<? extends T> other, 
Consumer<? super T> action
) {

引发错误的行如下所示：

.acceptEither(
timeoutAfter(50, TimeUnit.MILLISECONDS),
inPutItemGroup
)

因此，您会看到您尝试将ItemGroup作为推断VoidT的Consumer<? super T>传递，因此您得到预期的错误：

error: incompatible types: ItemGroup cannot be converted to Consumer<? super Void>