如何处理List<Observable<MyResponse>>？已尝试Observable.zip未得解

Got it, let's break this down for you. The issue with Observable.zip() you're probably hitting is that the common overloads take a fixed number of Observables as parameters, but your list is dynamic (any length). Luckily, RxJava has an overload of zip() that handles exactly this scenario—one that accepts an Iterable of Observables (which your ArrayList<Observable<MyResponse>> already is).

Core Solution: Use Observable.zip() with Iterable Overload #

This approach waits for all Observables in your list to complete, then combines their results into a single list of MyResponse objects. Perfect if you need to process all results together once every request finishes.

Step-by-Step Code Implementation #

First, here's how to adapt your existing code to use the correct zip() overload:

List<String> myList = myMethod();
List<Observable<MyResponse>> observables = new ArrayList<>();

// Populate your list of Observables as before
for (String value : myList) {
    observables.add(myAPI.getData(value));
}

// Zip all Observables into a single Observable<List<MyResponse>>
Observable.zip(observables, new Function<Object[], List<MyResponse>>() {
    @Override
    public List<MyResponse> apply(Object[] responses) throws Throwable {
        // Convert the Object array to a typed List<MyResponse>
        List<MyResponse> resultList = new ArrayList<>();
        for (Object response : responses) {
            resultList.add((MyResponse) response);
        }
        return resultList;
    }
})
.subscribe(new Observer<List<MyResponse>>() {
    @Override
    public void onSubscribe(Disposable d) {
        // Optional: Save the Disposable to cancel requests if needed
    }

    @Override
    public void onNext(List<MyResponse> myResponses) {
        // This is where you get all completed responses at once
        for (MyResponse response : myResponses) {
            // Process each individual MyResponse here
            // e.g., parse data, update UI, etc.
        }
    }

    @Override
    public void onError(Throwable e) {
        // Handle errors: if ANY Observable fails, this will trigger
        // e.g., log the error, show a user message
    }

    @Override
    public void onComplete() {
        // Triggered when all requests finish successfully
    }
});

Simplified with Lambda Expressions #

If you're using Java 8+, you can shorten this significantly:

Observable.zip(observables, responses -> {
    List<MyResponse> resultList = new ArrayList<>();
    for (Object res : responses) {
        resultList.add((MyResponse) res);
    }
    return resultList;
})
.subscribe(
    myResponses -> {
        // Process the full list of responses
        myResponses.forEach(response -> {
            // Individual response logic here
        });
    },
    error -> {
        // Error handling logic
    }
);

Handling Individual Request Failures #

By default, if any Observable in your list emits an error, the entire zip() operation will fail immediately. If you want to isolate failures (so one bad request doesn't break the whole batch), add error handling to each individual Observable:

for (String value : myList) {
    observables.add(myAPI.getData(value)
            // Return a default MyResponse if this request fails
            .onErrorReturn(throwable -> {
                // Log the error for debugging
                Log.e("API_ERROR", "Failed to get data for: " + value, throwable);
                // Return a placeholder or empty response
                return new MyResponse();
            })
            // Alternatively, use onErrorResumeNext to fallback to another Observable
            // .onErrorResumeNext(Observable.just(new MyResponse()))
    );
}

This way, even if one request fails, the zip() operation will still complete with all successful responses plus your default values for failed ones.

Alternative: Process Responses as They Complete #

If you don't need to wait for all requests to finish (e.g., you want to process each response as soon as it's ready), use Observable.concat() or Observable.merge() instead:

• concat(): Runs Observables in order (one after another), emits responses in the original list order.
• merge(): Runs Observables in parallel, emits responses in the order they complete.

Example with concat():

Observable.concat(observables)
        .subscribe(
            response -> {
                // Process each response immediately when it finishes
            },
            error -> {
                // Handle errors from individual requests
            },
            () -> {
                // Triggered when all requests are done
            }
        );

内容的提问来源于stack exchange，提问作者Narayanan P S