连续执行不同的 Completable答案

【问题标题】：Execute different Completables in succession连续执行不同的 Completable
【发布时间】：2017-01-20 10:32:24
【问题描述】：

我目前正在尝试使用 java 中的响应式扩展来实现特定的结果，但是我无法做到，也许你们中的某个人可以帮助我。

firstCompletable
  .onErrorComplete(t -> specificErrorHandlingOne())
  .andThen(secondCompletable())
  .onErrorComplete(t -> specificErrorHandlingTwo())
  .andThen(thirdCompletable())
  .onErrorComplete(t -> specificErrorHandlingThree())
  .andThen(fourthCompletable())
  .onErrorComplete(t -> specificErrorHandlingFour())
  .subscribe(viewCallback::showSuccess)

但是，当出现错误时，例如 secondCompletable，正在执行特定的错误处理，但其他 Completable 仍在调度中。如果其中一个 Completable 失败，我希望整个 Completable 链停止执行。我该怎么做？

我已经尝试使用 doOnError 代替，但这只是在抛出特定错误的堆栈跟踪上结束。

【问题讨论】：

标签： java rx-java reactivex

【解决方案1】：

Completable.concat(
    completable1.doOnError(e -> {...}),
    completable2.doOnError(e -> {...}),
    completable3.doOnError(e -> {...}),
    completable4.doOnError(e -> {...})
).subscribe(action, errorConsumer);

Completables 将按指定顺序订阅
全部完成后将调用action
您可以为每一个指定错误处理程序（这是可选的）
任何错误都会破坏管道并传播给订阅者 (errorConsumer)

您原来的andThen 链也应该可以工作，但您需要用doOnError 替换onErrorComplete，它用完成替换错误，它只是调用指定的操作。或者只是从您的specificErrorHandlingXxx() 返回false。

【讨论】：

嘿，看起来不错，但是当例如第二个 Completable 调用subscriber.onError(new Exception("Invalid phone number"))，这使我的Android 应用程序崩溃。正在调用 doOnError 处理方法，但随后它崩溃:(
@DavidBecher 那是因为错误被传播给订阅者，但订阅者不处理它。我已更新代码以添加此类错误使用者。
如果您的 sn-p 代码在一个方法 (P) 中，该方法本身返回 Completable，并且它本身也有 Completable.fromAction（它有两个 @987654331 @ methods inside(C) )，父方法(P) 是否等待那些Completable 子方法(C) 完成？
@Dr.jacky 方法是否等待某事取决于线程。如果事情在同一个线程上执行，那么它们会按顺序执行，否则同步是你的工作。

【解决方案2】：

试试下面的：

public static void main(String[] args) {
    System.out.println("start");
    Completable c1 = Completable.fromAction(() -> printAndWait(1, 1));
    Completable c2 = Completable.fromAction(() -> printAndWait(2, 2));
    Completable c3 = Completable.fromObservable(Observable.timer(3, TimeUnit.SECONDS).concatWith(Observable.error(new RuntimeException())));
    Completable c4 = Completable.fromAction(() -> printAndWait(4, 2));

    c1.concatWith(c2).concatWith(c3).concatWith(c4).subscribe(e -> e.printStackTrace(), () -> System.out.println("done"));

    printAndWait(10, 10);//dont exit till program is completely executed

}

private static void printAndWait(int i, int j) {
    System.out.println(i);
    Observable.timer(j, TimeUnit.SECONDS).toBlocking().subscribe();//just add delay
}

【讨论】：