java - RxJava 可完成连接

Question

考虑以下代码片段:

@Test
public void completableConcatTest() {
    Completable.concat(completableTwoSeconds(), completableTwoSeconds())
            .doOnCompleted(() -> System.out.println("Both completed."))
            .await();
}

private Completable completableTwoSeconds() {
    PublishSubject<Void> subject = PublishSubject.create();

    CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        subject.onCompleted();
        return null;
    });

    return subject.doOnCompleted(() -> System.out.println("I'm completed." + Instant.now())).toCompletable();
}

运行时，测试会产生以下输出:

I'm completed.2018-05-11T10:07:26.359Z
I'm completed.2018-05-11T10:07:26.380Z
Both completed.

显然，两个可完成项是并行运行的，而不是按预期串行运行。

你能解释一下我在这里做错了什么吗？

提前致谢!

Answer 1

CompletableFuture 几乎在声明后立即开始执行，您可以在 RxJava 和 concat 参与之前以 completableTwoSeconds() 方式执行此操作。您必须至少推迟对 completableTwoSeconds() 的第二次调用，以便在第一次完成后执行它:

Completable.concat(
    completableTwoSeconds(), 
    Completable.defer(() -> completableTwoSeconds())
)
.doOnCompleted(() -> System.out.println("Both completed."))
.await();