如何在 RxJava2 中链接两个 Completable

Question

我有两个 Completable。我想做以下场景： 如果第一个 Completable 到达 onComplete ，则继续第二个 Completable。最终结果将是第二个 Completable 的 onComplete。

当我有 Single getUserIdAlreadySavedInDevice() 和 Completable login() 时，我就是这样做的：

@Override
public Completable loginUserThatIsAlreadySavedInDevice(String password) {
    return getUserIdAlreadySavedInDevice()
            .flatMapCompletable(s -> login(password, s))

}

Answer 1

您正在寻找andThen 运营商。

返回一个 Completable，它首先运行这个 Completable，然后运行另一个 Completable。

firstCompletable
    .andThen(secondCompletable)

通常，此运算符是Completable 上flatMap 的“替换”：

Completable       andThen(CompletableSource next)
<T> Maybe<T>      andThen(MaybeSource<T> next)
<T> Observable<T> andThen(ObservableSource<T> next)
<T> Flowable<T>   andThen(Publisher<T> next)
<T> Single<T>     andThen(SingleSource<T> next)

Answer 2

TL;DR：其他答案错过了一个微妙之处。如果你想要 concat 的等价物，请使用 doThingA().andThen(doThingB())，如果你想要 flatMap 的等价物，请使用 doThingA().andThen(Completable.defer(() -> doThingB())。

---

编辑：更完整的参考

• flatMap()是merge()的映射版本
• concatMap() 是concat() 的映射版本
• 对于Completable，您需要defer() 来使函数调用变得惰性，就像在Single 或Observable 的映射函数中一样（或者最好让它在您点击订阅之前什么都不会发生 - 这是一个很好的约定遵循并在官方 Rx 库以及我遇到的任何 Rx 扩展中使用，对于高级用户，这仅指冷可完成，但大多数人可以忽略这一点。
• concat(a, b) 和 a.andThen(b) 之间的唯一区别是语法

一些例子：

• foo(a).andThen(bar(b)) 将：

  1. 致电foo(a)
  2. 立即调用bar(b)，即使步骤1返回的可完成返回错误
  3. 订阅1 返回的任何步骤
  4. 然后将订阅bar(b)的结果仅当最后一步成功完成时

• foo(a).andThen(Completable.defer(() -> bar(b)) 将：

  1. 致电foo(a)
  2. 订阅步骤1的结果
  3. 只有当foo(a)返回的可完成函数成功时，才会调用bar(b)

我将省略merge() 的处理，因为它有点复杂，但长话短说，如果你想要“并行性”，那就是调用它。

---

上述答案有点正确，但我发现它们具有误导性，因为它们忽略了急切评估的微妙之处。

doThingA().andThen(doThingB()) 将立即调用 doThingB()，但仅在 doThingA() 返回的 observable 完成时订阅 doThingB() 返回的 observable。

doThingA().andThen(Completable.defer(() -> doThingB()) 只有在事情 A 完成后才会调用 doThingB()。

仅当doThingB() 在订阅事件之前有副作用时，这才重要。例如。 Single.just(sideEffect(1)).toCompletable()

在订阅事件（真正的冷可观察）之前没有副作用的实现可能是Single.just(1).doOnSuccess(i -> sideEffect(i)).toCompletable()。

如果这让我很生气，A 是一些验证逻辑，doThingB() 会立即启动异步数据库更新，从而完成 VertX ObservableFuture。这是不好的。可以说doThingB() 应该写成只在订阅时更新数据库，我将在未来尝试以这种方式设计。

Answer 3

试试

Completable.concat

Returns a Completable which completes only when all sources complete, one after another.

http://reactivex.io/RxJava/javadoc/io/reactivex/Completable.html#concat(java.lang.Iterable)

Answer 4

请注意，这里获得更多选票的答案有点误导。看下面的例子，我的想法是展示一些测试场景，并展示带有运算符andThen 的可完成逻辑的行为。

 private fun doThingAFail(): Completable {
        print("Do thingA Called\n")
        return Completable.fromCallable {
            print("calling stream A\n")
            throw(Exception("The excep"))
        }
    }

    private fun doThingB(): Completable {
        print("Do thingB Called\n")
        return Completable.fromCallable {
            print("calling stream B\n")

        }
    }

    private fun doThingA(): Completable {
        print("Do thingA Called\n")
        return Completable.fromCallable {
            print("calling stream A\n")
        }
    }

请注意以下测试：

@Test
fun testCallAPlusB() {
    doThingA().andThen(doThingB())
}

输出将是：

Do thingA Called
Do thingB Called

此处快速说明：请注意，我们没有在此 sn-p 中订阅这些 Completable。

为了测试：

@Test
fun theTestSubscribe() {
    doThingA().andThen(doThingB()).subscribe()
}

输出将是：

Do thingA Called
Do thingB Called
calling stream A
calling stream B

最后，如果第一个 Completable 失败，第二个 Completable 将不会被执行。

@Test
fun theTestFailThingA() {
    doThingAFail().andThen(doThingB()).subscribe()
}

输出将是：

Do thingA Called
Do thingB Called
calling stream A

这里的关键概念是方法内部和可观察对象内部的逻辑不是同时执行的。 一旦我们调用 doThingA() 和 doThingB() 方法，将打印“Do thingA Called”和“Do thingB Called”行。而“调用流A”和 只有当有人订阅 doThingA 和 doThingB 方法时，才会调用“调用流 B”行。

这里的第二个概念是 andThen 运算符将如何处理错误。在上面的示例中，如果doThingA() completable 以错误结束，流将结束并且不打印“调用流 B”行。

Answer 5

我遇到了同样的问题，我使用了操作符 .concactWith 来使它工作。 就我而言，我有两种 Completable 类型的乐趣。

fun makeTwoThings(): Completable {
    makeFirstThing().concatWith(makeSecondThing())
}

fun makeFirstThing(): Completable{
     //TODO()
}

fun makeSecondThing(): Completable{
     //TODO()
}