我有一系列异步服务调用,我想取消它。好吧,实际上,我有两条服务调用链并行进行,如果其中一个成功,我想取消另一个。
对于番石榴的期货,我习惯于通过取消最后一个期货来取消整个期货链。看来我无法用 java-8 的 future 做到这一点。 除非有人知道如何做。
你的任务,如果你选择接受它,就是告诉我是否可以保留我漂亮的语法并取消链。否则,我将编写自己的链接未来包装器 - 特别是在这个问题之后。
我自己的测试和尝试如下。
@Test
public void shouldCancelOtherFutures() {
// guava
ListenableFuture<String> as = Futures.immediateFuture("a");
ListenableFuture<String> bs = Futures.transform(as, (AsyncFunction<String, String>) x -> SettableFuture.create());
ListenableFuture<String> cs = Futures.transform(bs, (AsyncFunction<String, String>) x -> SettableFuture.create());
ListenableFuture<String> ds = Futures.transform(cs, Functions.<String>identity());
ds.cancel(false);
assertTrue(cs.isDone()); // succeeds
// jdk 8
CompletableFuture<String> ac = CompletableFuture.completedFuture("a");
CompletableFuture<String> bc = ac.thenCompose(x -> new CompletableFuture<>());
CompletableFuture<String> cc = bc.thenCompose(x -> new CompletableFuture<>());
CompletableFuture<String> dc = cc.thenApply(Function.identity());
dc.cancel(false);
assertTrue(cc.isDone()); // fails
}
(想象每个
thenCompose()
和 Futures.transform(x, AsyncFunction)
代表一个异步服务调用。)
我明白为什么道格·李的研究生大军会这样做。有了分支链,一切都应该取消吗?
CompletableFuture<Z> top = new CompletableFuture<>()
.thenApply(x -> y(x))
.thenCompose(y -> z(y));
CompletableFuture<?> aBranch = top.thenCompose(z -> aa(z));
CompletableFuture<?> bBranch = top.thenCompose(z -> bb(z));
...
bBranch.cancel(false);
// should aBranch be canceled now?
我可以使用自定义包装函数来解决这个问题,但它弄乱了漂亮的语法。
private <T,U> CompletableFuture<U> transformAsync(CompletableFuture<T> source, Function<? super T,? extends CompletableFuture<U>> transform) {
CompletableFuture<U> next = source.thenCompose(transform);
next.whenComplete((x, err) -> next.cancel(false));
return next;
}
private <T,U> CompletableFuture<U> transform(CompletableFuture<T> source, Function<T,U> transform) {
CompletableFuture<U> next = source.thenApply(transform);
next.whenComplete((x, err) -> next.cancel(false));
return next;
}
// nice syntax I wished worked
CompletableFuture<?> f1 = serviceCall()
.thenApply(w -> x(w))
.thenCompose(x -> serviceCall())
.thenCompose(y -> serviceCall())
.thenApply(z -> $(z));
// what works, with less readable syntax
CompletableFuture<?> f2 =
transform(
transformAsync(
transformAsync(
transform(serviceCall, x(w)),
x -> serviceCall()),
y -> serviceCall()),
z -> $(z));
这取决于您的目标是什么。我认为,让中间
CompletableFuture
报告完成状态并不重要,因为在使用链式构造调用时您通常不会注意到。重要的一点是,您希望您昂贵的serviceCall()
不要被触发。
一种解决方案可以是:
CompletableFuture<String> flag=new CompletableFuture<>();
CompletableFuture<String> ac = serviceCall()
.thenCompose(x -> flag.isCancelled()? flag: serviceCall())
.thenCompose(x -> flag.isCancelled()? flag: serviceCall());
ac.whenComplete((v,t)->flag.cancel(false));// don’t chain this call
这使用像解决方案中那样的
whenComplete
调用,但仅在最后的 CompletableFuture
上将取消传播到专用 flag
对象。调用 cancel
后,下一个 thenCompose
调用将检测取消并返回取消的 future,因此取消将传播链,因此不再调用 compose 或 apply 方法。
缺点是它不能与
thenApply
结合使用,因为 Function
无法返回取消的 future。因此,当异步服务调用完成并与 Function
链接时,即使发生取消,也会应用该函数。
解决此问题的替代解决方案是为您的
serviceCall
创建一个包装函数,其中包括启动前和完成后的测试:
CompletableFuture<String> serviceCall(CompletableFuture<String> f) {
if(f.isCancelled()) return f;
CompletableFuture<String> serviceCall=serviceCall();
return serviceCall.thenCompose(x->f.isCancelled()? f: serviceCall);
}
那么您的用例将如下所示:
CompletableFuture<String> flag=new CompletableFuture<>();
CompletableFuture<String> ac = serviceCall(flag)
.thenApply(w->x(w))
.thenCompose(x -> serviceCall(flag))
.thenCompose(x -> serviceCall(flag))
.thenApply(z -> $(z));
ac.whenComplete((v,t)->flag.cancel(false));
当然,您必须将
<String>
替换为原始 serviceCall()
用于 CompletableFuture<T>
的任何类型参数。
您可以尝试 futures4j 库的 ExtendedFuture 类,它扩展了
CompletableFuture
并支持反向链取消,即取消依赖阶段会取消父阶段。请参阅https://github.com/futures4j/futures4j#ExtendedFuture
var myFuture = ExtendedFuture.supplyAsync(...);
var myCancellableChain = myFuture
.thenApply(...) // Step 1
.asCancellableByDependents(true) // allows subsequent stages cancel the previous stage
.thenRun(...) // Step 2
.thenApply(...) // Step 3
.thenAccept(...); // Step 4
if (shouldCancel) {
// Cancels steps 4 to 1
myCancellableChain.cancel(true);
}