网站流量指标有哪些,东营人才网,wordpress在哪里,响应式网站建设流程Java 8中的CompletableFutureT是对T类型的值将来将可用的承诺的高级抽象。 ObservableT非常相似#xff0c;但是它承诺将来会出现任意数量的项#xff0c;从0到无穷大。 异步结果的这两种表示与仅使用一项即可使用Observable而不是CompletableFuture情况非常… Java 8中的CompletableFutureT是对T类型的值将来将可用的承诺的高级抽象。 ObservableT非常相似但是它承诺将来会出现任意数量的项从0到无穷大。 异步结果的这两种表示与仅使用一项即可使用Observable而不是CompletableFuture情况非常相似反之亦然。 另一方面 CompletableFuture更专业并且由于它现在是JDK的一部分因此应该很快就会流行起来。 让我们用简短的文章来庆祝RxJava 1.0发行版该文章展示了如何在两者之间进行转换而又不失去它们的异步和事件驱动性质。 从 CompletableFuture表示将来的一个值因此将其变为Observable非常简单。 当Future以某个值完成时 Observable也将立即发出该值并关闭流 class FuturesTest extends Specification {public static final String MSG Dont panicdef should convert completed Future to completed Observable() {given:CompletableFutureString future CompletableFuture.completedFuture(Abc)when:ObservableString observable Futures.toObservable(future)then:observable.toBlocking().toIterable().toList() [Abc]}def should convert failed Future into Observable with failure() {given:CompletableFutureString future failedFuture(new IllegalStateException(MSG))when:ObservableString observable Futures.toObservable(future)then:observable.onErrorReturn({ th - th.message } as Func1).toBlocking().toIterable().toList() [MSG]} CompletableFuture failedFuture(Exception error) {CompletableFuture future new CompletableFuture()future.completeExceptionally(error)return future}} 尚未执行的 Futures.toObservable()第一个测试会将Future转换为Observable 并确保正确传播值。 第二个测试创建了失败的Future 将失败替换为异常的消息并确保传播了异常。 实现要短得多 public static T ObservableT toObservable(CompletableFutureT future) {return Observable.create(subscriber -future.whenComplete((result, error) - {if (error ! null) {subscriber.onError(error);} else {subscriber.onNext(result);subscriber.onCompleted();}}));
} 注意 Observable.fromFuture()存在但是我们想充分利用ComplatableFuture的异步运算符。 从 实际上有两种将Observable转换为Future -创建CompletableFutureListT或CompletableFutureT 如果我们假设Observable仅包含一项。 让我们从前一种情况开始用以下测试用例进行描述 def should convert Observable with many items to Future of list() {given:ObservableInteger observable Observablejust(1, 2, 3)when:CompletableFutureListInteger future FuturesfromObservable(observable)then:futureget() [1, 2, 3]
}def should return failed Future when after few items exception was emitted() {given:ObservableInteger observable Observablejust(1, 2, 3)concatWith(Observableerror(new IllegalStateException(MSG)))when:FuturesfromObservable(observable)then:def e thrown(Exception)emessage MSG
} 显然直到源Observable信号流结束 Future才完成。 因此 Observable.never()将永远不会完成包装Future 而是使用空列表来完成它。 该实现更短更甜蜜 public static T CompletableFutureListT fromObservable(ObservableT observable) {final CompletableFutureListT future new CompletableFuture();observable.doOnError(future::completeExceptionally).toList().forEach(future::complete);return future;
} 关键是Observable.toList() 它可以方便地从ObservableT和ObservableListT 。 当源ObservableT完成时后者将发出ListT类型的一项。 从 当我们知道CompletableFutureT将恰好返回一项时就会发生上一次转换的特殊情况。 在这种情况下我们可以将其直接转换为CompletableFutureT 而不是仅包含一项的CompletableFutureListT 。 首先测试 def should convert Observable with single item to Future() {given:ObservableInteger observable Observable.just(1)when:CompletableFutureInteger future Futures.fromSingleObservable(observable)then:future.get() 1
}def should create failed Future when Observable fails() {given:ObservableString observable Observable.String error(new IllegalStateException(MSG))when:Futures.fromSingleObservable(observable)then:def e thrown(Exception)e.message MSG
}def should fail when single Observable produces too many items() {given:ObservableInteger observable Observable.just(1, 2)when:Futures.fromSingleObservable(observable)then:def e thrown(Exception)e.message.contains(too many elements)
} 同样实现非常简单并且几乎相同 public static T CompletableFutureT fromSingleObservable(ObservableT observable) {final CompletableFutureT future new CompletableFuture();observable.doOnError(future::completeExceptionally).single().forEach(future::complete);return future;
} 上面的Helper方法还不够完善但是如果您需要在JDK 8和RxJava风格的异步计算之间进行转换那么这篇文章应该足以帮助您入门。 翻译自: https://www.javacodegeeks.com/2014/12/converting-between-completablefuture-and-observable.html
