java - 避免 RxJava 中的 toBlocking

Question

我发现下面的代码(为简洁起见进行了修改)，它有错误，并且据我所知存在缺陷。

1. 它使用 toBlocking() ，一般不建议使用，并且已在响应式(Reactive)上下文中使用
2. 它返回 single();因此，一旦它从流中获取单个项目，它就会终止它，忽略其余不需要的项目。

虽然我相信我可以通过将 single 删除到 last 来解决问题？有人可以解释一下，我怎样才能在这里删除 toBlocking() 的使用？

observableList.map(incentiveDetailsList -> {
        List<SomeObject> list = mapThisList(incentiveDetailsList);
        return Observable.just(list)
                    .flatMap(Observable::from)
                    .flatMap(item -> {
                        Request request = createRequest(item);
                        String accountNumber = item.getAccountNumber();
                        return serviceThatReturnsObservable.load(request)
                                .doOnError(onError -> {
                                    Observable.error(new Exception("some context"));
                                })
                                .map(response -> {
                                    handleError(response);
                                    return responseMap.put(accountNumber, buildResponse(response.getResponse()));
                                });
                    })
                    .map(resp -> mapResponse(store, incentiveDetailsList, responseMap))
                    .toBlocking()
                    .single();
})

Answer 1

您可以用flatMap(或concatMap)替换map + toBlocking:

 observableList.flatMap(incentiveDetailsList -> {
    List<SomeObject> list = mapThisList(incentiveDetailsList);
    return Observable.from(list)
           .flatMap(item -> {
               Request request = createRequest(item);
               String accountNumber = item.getAccountNumber();
               return serviceThatReturnsObservable.load(request)
                   /* this has no effect:
                            .doOnError(onError -> {
                                Observable.error(new Exception("some context"));
                            })
                    */
                    .map(response -> {
                         handleError(response);
                         return responseMap.put(accountNumber, 
                            buildResponse(response.getResponse()));
                    });
            })
            .map(resp -> mapResponse(store, incentiveDetailsList, responseMap));
})