如何展平 Single<R> 流

Question

我有这个代码    public Stream<PaymentEntity> getPayments(List<String> paymentIds) {        return paymentIds.stream()                .flatMap(id -> paymentsRepository.getById(id))}paymentsRepository.getById(id)正在返回Single<PaymentEntity>但我遇到编译时错误不存在类型变量 R 的实例，因此 Single 符合 Stream

Answer 1

您可以使用

return paymentIds.stream()
            .map(id -> paymentsRepository.getById(id).toBlocking().value());

或者rxjava 2您可以使用：

 return paymentIds.stream()
            .map(id -> paymentsRepository.getById(id).blockingGet());

---

评论后我认为你可以采用这个解决方案：

 List<PaymentEntity> result = new ArrayList<>();
 paymentIds.forEach(id -> paymentsRepository.getById(id).toObservable().subscribe(result::add));
  return result.stream();

Answer 2

替代解决方案：

public Stream<PaymentEntity> getPayments(List<String> paymentIds) {
    return Observable.fromIterable(paymentIds)
            .flatMapSingle(id -> paymentsRepository.getById(id))
            .toList()
            .blockingGet()
            .stream();
}

Answer 3

您可以等待，直到Singlelambda 中的当前信号发出成功值，使用.blockingGet()：

return paymentIds.stream()
            .map(id -> paymentsRepository.getById(id).blockingGet());

或者您可以等待，直到所有反应链都以这种方式发出成功值：

return Observable.fromIterable(paymentIds)
            .flatMapSingle(paymentsRepository::getById)
            .toList()
            .blockingGet()
            .stream();

还可以考虑使用Flowable：与 不同的是Observable，它支持背压策略。