我有一个Set< Object>对于Set中的每个条目,我必须进行API调用,将其作为参数传递.而且我必须处理每个响应和&使用自己的逻辑填充另一个Map

示例顺序执行：

List<MyResponse> responses = newArrayList<>();
Set<StoreNode> nodes = // Assume we have a Set
nodes.forEach(storeNode -> responses.add(myAPI.myMethod(storeNode.getId()));
responses.forEach(response -> processResponse(response, myMap); // This is a common map & I have some custom logic to populate this map

如何使用Observables实现相同的目标？我想和平地做这些电话&填充我的公共地图myMap

我遇到了map(),flatMap()& zip()但我看到的大多数例子都是简单的,没有进行API调用和处理他们的回应.

解决方法:

这取决于您使用的是哪个RxJava版本.如果它早于2.0.5,那么你需要做flatMap,在那里你创建另一个Observable并确保那里的东西是pallel.请参阅*上的this answer.

否则,我建议使用Flowable,然后您可以使用parallel()运算符将Flowable更改为ParallelFlowable.

所以你可以这样做：

Flowable.fromIterable(nodes)
        .parallel() // you can also specify number of rails here
        .runOn(Schedulers.computation())
        .map(node -> myAPI.myMethod(node.getId()))
        .sequential()
        .subscribe(
                response -> processResponse(response, myMap),
                error -> log(error)
        );

有关更多信息,请参阅Parallel flows文档.