你的问题是 switchMapDelayError 接线员。您应该使用concatmap或flatmap
我冒昧地为您的用例编写了一个测试。请注意，始终使用重载来提供 Scheduler 为了提供 TestScheduler 用于测试。

switchmap做什么？

在每个上游emit switchmap上订阅给定的内部流。当一个新的值从上游发出时，旧的内部流被取消订阅，并且switchmap的lambda被再次调用以订阅新的内部流。
问题可能是以下代码：

return Observable
            .just(
                    new Response(1),
                    new Response(2),
                    new Response(3),
                    new Response(4))
            .delay(100, TimeUnit.MILLISECONDS);

它几乎一个接一个地在堆栈上发出响应1到4，并且每个发出都在另一个线程上延迟。因此，响应1到4几乎会立即发出。它们不会像以下那样发出：100ms时的响应（1），200ms时的响应（2），等等。
让我们看看输出是为了什么

Observable.just(
    new Response(1), //
    new Response(2),
    new Response(3),
    new Response(4))
    .delay(100, TimeUnit.MILLISECONDS)
    .subscribe(r -> {
      System.out.println("received value at " + Schedulers.io().now(TimeUnit.MILLISECONDS));
    });

输出

received value at 1607432032768
received value at 1607432032769
received value at 1607432032769
received value at 1607432032769

因此，所有值几乎都会立即发出，并用switchmap相互覆盖。以前发出的值几乎立即被新值抵消。

解决方案

使用concatmap或flatmap或更改测试设置以100ms的间隔发出每个值。
flatmap只订阅每个值，默认情况下最多128个内部流。当内部流完成时，concatmap将只订阅下一个值。
测试

public class So65193002 {
      @Test
      void so() {
        TestScheduler testScheduler = new TestScheduler();
        Service1 service1 = new Service1(testScheduler);
        Service2 service2 = new Service2(testScheduler);

        Observable<Response> service1Call = service1.call(1);

        Observable<Response> combined =
            service1Call.concatMapEagerDelayError(
                x -> {
                  return service2
                      .call(x.value)
                      .onErrorReturn(
                          (ex) -> {
                            x.warning = ex.getMessage();
                            return x; // returns at least service1 result
                          });
                },
                true);

        TestObserver<Response> test = combined.test();

        testScheduler.advanceTimeBy(1, TimeUnit.HOURS);

        test.assertValueCount(4)
            .assertValueAt(
                0,
                r -> {
                  assertThat(r.value).isEqualTo(1);
                  assertThat(r.warning).isNotEmpty();
                  return true;
                })
            .assertValueAt(
                1,
                r -> {
                  assertThat(r.value).isEqualTo(200);
                  assertThat(r.warning).isNull();
                  return true;
                })
            .assertValueAt(
                3,
                r -> {
                  assertThat(r.value).isEqualTo(400);
                  assertThat(r.warning).isNull();
                  return true;
                });
      }
    }

域

class Response {
  public Integer value;
  public String warning;

  public Response(Integer value) {
    this.value = value;
  }

  @Override
  public String toString() {
    return "Response{" + "value=" + value + ", warning='" + warning + '\'' + '}';
  }
}

class Service1 {
  private final Scheduler scheduler;

  Service1(Scheduler scheduler) {
    this.scheduler = scheduler;
  }

  public Observable<Response> call(int arg) {
    return Observable.just(
            new Response(1), //
            new Response(2),
            new Response(3),
            new Response(4))
        .delay(100, TimeUnit.MILLISECONDS, scheduler);
  }
}

class Service2 {
  private final Scheduler scheduler;

  Service2(Scheduler scheduler) {
    this.scheduler = scheduler;
  }

  public Observable<Response> call(int arg) {
    if (arg % 2 == 0) {
      return Observable.just(new Response(100 * arg)).delay(10, TimeUnit.MILLISECONDS, scheduler);

    } else {
      return Observable.error(new RuntimeException("service 2 error"));
    }
  }
}

注意

不要使用可变对象。始终确保发出的值是不可变的，否则会遇到麻烦。