本文整理了Java中io.reactivex.Flowable.flatMapSingle()方法的一些代码示例，展示了Flowable.flatMapSingle()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Flowable.flatMapSingle()方法的具体详情如下：
包路径：io.reactivex.Flowable
类名称：Flowable
方法名：flatMapSingle

Flowable.flatMapSingle介绍

[英]Maps each element of the upstream Flowable into SingleSources, subscribes to all of them and merges their onSuccess values, in no particular order, into a single Flowable sequence. Backpressure: The operator consumes the upstream in an unbounded manner. Scheduler: flatMapSingle does not operate by default on a particular Scheduler.
[中]将上游可流动的每个元素映射到单个源，订阅所有这些元素，并将它们的onSuccess值（无特定顺序）合并到单个可流动序列中。背压：操作员以无限制的方式消耗上游压力。调度程序：默认情况下，flatMapSingle不会在特定调度程序上运行。

代码示例

代码示例来源：origin: ReactiveX/RxJava

/**
 * Maps each element of the upstream Flowable into SingleSources, subscribes to all of them
 * and merges their onSuccess values, in no particular order, into a single Flowable sequence.
 * <dl>
 *  <dt><b>Backpressure:</b></dt>
 *  <dd>The operator consumes the upstream in an unbounded manner.</dd>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code flatMapSingle} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 * @param <R> the result value type
 * @param mapper the function that received each source value and transforms them into SingleSources.
 * @return the new Flowable instance
 */
@CheckReturnValue
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Flowable<R> flatMapSingle(Function<? super T, ? extends SingleSource<? extends R>> mapper) {
  return flatMapSingle(mapper, false, Integer.MAX_VALUE);
}

代码示例来源：origin: ReactiveX/RxJava

@Override
  public Flowable<Integer> apply(Flowable<Object> f) throws Exception {
    return f.flatMapSingle(Functions.justFunction(Single.just(2)));
  }
});

代码示例来源：origin: ReactiveX/RxJava

@Test
public void normalDelayError() {
  Flowable.range(1, 10)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(v);
    }
  }, true, Integer.MAX_VALUE)
  .test()
  .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void normal() {
  Flowable.range(1, 10)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(v);
    }
  })
  .test()
  .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void middleError() {
  Flowable.fromArray(new String[]{"1", "a", "2"})
  .flatMapSingle(new Function<String, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(final String s) throws NumberFormatException {
      //return Single.just(Integer.valueOf(s)); //This works
      return Single.fromCallable(new Callable<Integer>() {
        @Override
        public Integer call() throws NumberFormatException {
          return Integer.valueOf(s);
        }
      });
    }
  })
  .test()
  .assertFailure(NumberFormatException.class, 1);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void normalMaxConcurrent2Backpressured() {
  Flowable.range(1, 10)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(v);
    }
  }, false, 2)
  .rebatchRequests(1)
  .test()
  .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void normalMaxConcurrent1Backpressured() {
  Flowable.range(1, 10)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(v);
    }
  }, false, 1)
  .rebatchRequests(1)
  .test()
  .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void normalAsyncMaxConcurrency1() {
  Flowable.range(1, 10)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(v).subscribeOn(Schedulers.computation());
    }
  }, false, 1)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void normalBackpressured() {
  Flowable.range(1, 10)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(v);
    }
  })
  .rebatchRequests(1)
  .test()
  .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void take() {
  Flowable.range(1, 10)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(v);
    }
  })
  .take(2)
  .test()
  .assertResult(1, 2);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void normalAsync() {
  Flowable.range(1, 10)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(v).subscribeOn(Schedulers.computation());
    }
  })
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertSubscribed()
  .assertValueSet(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
  .assertNoErrors()
  .assertComplete();
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void normalAsyncMaxConcurrency() {
  Flowable.range(1, 10)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(v).subscribeOn(Schedulers.computation());
    }
  }, false, 3)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertSubscribed()
  .assertValueSet(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
  .assertNoErrors()
  .assertComplete();
 }

代码示例来源：origin: ReactiveX/RxJava

@Test
public void error() {
  Flowable.just(1)
  .flatMapSingle(Functions.justFunction(Single.<Integer>error(new TestException())))
  .test(0L)
  .assertFailure(TestException.class);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void errorDelayed() {
  Flowable.just(1)
  .flatMapSingle(Functions.justFunction(Single.<Integer>error(new TestException())), true, 16)
  .test(0L)
  .assertFailure(TestException.class);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void backpressure() {
  TestSubscriber<Integer> ts = Flowable.just(1)
  .flatMapSingle(Functions.justFunction(Single.just(2)))
  .test(0L)
  .assertEmpty();
  ts.request(1);
  ts.assertResult(2);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void takeAsync() {
  Flowable.range(1, 10)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(v).subscribeOn(Schedulers.computation());
    }
  })
  .take(2)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertSubscribed()
  .assertValueCount(2)
  .assertValueSet(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
  .assertNoErrors()
  .assertComplete();
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void successError() {
  final PublishProcessor<Integer> pp = PublishProcessor.create();
  TestSubscriber<Integer> ts = Flowable.range(1, 2)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      if (v == 2) {
        return pp.singleOrError();
      }
      return Single.error(new TestException());
    }
  }, true, Integer.MAX_VALUE)
  .test();
  pp.onNext(1);
  pp.onComplete();
  ts
  .assertFailure(TestException.class, 1);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void normalDelayErrorAll() {
  TestSubscriber<Integer> ts = Flowable.range(1, 10).concatWith(Flowable.<Integer>error(new TestException()))
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.error(new TestException());
    }
  }, true, Integer.MAX_VALUE)
  .test()
  .assertFailure(CompositeException.class);
  List<Throwable> errors = TestHelper.compositeList(ts.errors().get(0));
  for (int i = 0; i < 11; i++) {
    TestHelper.assertError(errors, i, TestException.class);
  }
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void innerSuccessCompletesAfterMain() {
  PublishProcessor<Integer> pp = PublishProcessor.create();
  TestSubscriber<Integer> ts = Flowable.just(1).flatMapSingle(Functions.justFunction(pp.singleOrError()))
  .test();
  pp.onNext(2);
  pp.onComplete();
  ts
  .assertResult(2);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void asyncFlatten() {
  Flowable.range(1, 1000)
  .flatMapSingle(new Function<Integer, SingleSource<Integer>>() {
    @Override
    public SingleSource<Integer> apply(Integer v) throws Exception {
      return Single.just(1).subscribeOn(Schedulers.computation());
    }
  })
  .take(500)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertSubscribed()
  .assertValueCount(500)
  .assertNoErrors()
  .assertComplete();
}