如何让 RetryWhen 重试抛出异常的流

Question

我正在尝试使用retryWhen 处理一个非常基本的流程。 我正在发射 3 个 Flowable，其中一个会抛出一个 IOException，在这种情况下，我最多会触发 2 次重试。 问题是重试时它会重新启动所有内容..导致其他流重新发射。 这是我的代码：

Flowable.just("AA", "BB", "CC")//
        .flatMap(station -> getStation(station))//
        .retryWhen( RetryWhen
                     .maxRetries(2)
                     .retryWhenInstanceOf(IOException.class)
                     .build())
        .subscribe(//
            station -> System.out.println("Received Availability for station=" + station),
            error -> System.err.println("Failed with error=" + error.getMessage()),
            () -> System.out.println("Completed!")//
        );

private Flowable<String> getStation(String station)
  {
    if (station.equals("CC"))
    {
      System.err.println("Failed staton=" + station + " --> Going to retry");
      return Flowable.error(new IOException("Server for station=" + station + " is down!"));
    }

    System.out.println("Querying for Station=" + station);

    return Flowable.just(station);
  }

如何调整它以仅使抛出异常的那个重试？？

编辑： 根据反馈，我已更改代码以在每个Flowable 实例上重试：

 Flowable<String> flw1 = getStationAvailability("AA");
        Flowable<String> flw2 = getStationAvailability("BB");
        Flowable<String> flw3 = getStationAvailability("CC");

      Flowable.concat(//
            flw1.retryWhen(RetryWhen.maxRetries(2).retryWhenInstanceOf(IOException.class).build()),
            flw2.retryWhen(RetryWhen.maxRetries(2).retryWhenInstanceOf(IOException.class).build()),
            flw3.retryWhen(RetryWhen.maxRetries(2).retryWhenInstanceOf(IOException.class).build())//
        ).subscribe(//
            station -> System.out.println("Received Availability for station=" + station),
            error -> System.err.println("Failed with error=" + error.getMessage()),// 
            () -> System.out.println("Completed!")//
        );

但是，发生的情况是它根本不重试。 对此有何见解？ 谢谢！

Answer 1

您需要将retryWhen() 运算符放在各个可观察对象的观察者链上。

Flowable.just("AA", "BB", "CC")//
        .flatMap(station -> getStation(station)
          .retryWhen( retryStrategy ) )//
      .subscribe( ... );

这样，重新订阅只发生在一个观察者链上。