我是否通过将一个可观察对象转换为阻塞可观察对象而误用了rxJava？

Question

我的API对两个独立的服务进行大约100次下游调用。在我将回复返回给客户之前，所有回复都需要汇总。我使用hystrix-feign进行HTTP调用。

我提出了一个我认为是优雅的解决方案，直到在rxJava文档中我发现了以下内容

BlockingObservable是提供阻塞运算符的各种可观察对象。它可以用于测试和演示目的，但通常不适用于生产应用程序（如果您认为需要使用BlockingObservable，这通常是您应该重新考虑设计的标志）。

我的代码大致如下

List<Observable<C>> observables = new ArrayList<>();
for (RequestPair request : requests) {
    Observable<C> zipped = Observable.zip(
         feignClientA.sendRequest(request.A()),
         feignClientB.sendRequest(request.B()),
         (a, b) -> new C(a,b));
    observables.add(zipped);
}

Collection<D> apiResponse = = new ConcurrentLinkedQueue<>();

Observable
    .merge(observables)
    .toBlocking()
    .forEach(combinedResponse -> apiResponse.add(doSomeWork(combinedResponse)));

return apiResponse;

基于此设置的几个问题：

1. 考虑到我的用例，toBlocking（）是否合理

Answer 1

一个更好的选择是返回可观察的，供其他操作员使用，但您可以通过阻塞代码（不过，它应该在后台线程上运行）

public Observable<D> getAll(Iterable<RequestPair> requests) {
    return Observable.from(requests)
    .flatMap(request ->
        Observable.zip(
            feignClientA.sendRequest(request.A()),
            feignClientB.sendRequest(request.B()),
            (a, b) -> new C(a,b)
        )
    , 8)  // maximum concurrent HTTP requests
    .map(both -> doSomeWork(both));
}

// for legacy users of the API
public Collection<D> getAllBlocking(Iterable<RequestPair> requests) {
    return getAll(requests)
        .toList()
        .toBlocking()
        .first();
}

我的理解正确吗，在主线程到达forEach（）之前，不会进行实际的HTTP调用

是的，forEach触发整个操作序列。

我已经看到forEach（）块中的代码由不同的线程执行，但我无法验证forEach（）块中是否可以有多个线程。那里的处决是同时进行的吗？

每次只允许一个线程在forEach中执行lambda，但您可能确实看到不同的线程进入其中。