RxJava中ConcatMap无跳过错误选项，序列下载失败如何跳转至下一个？

Hey there! I totally get where you're stuck—this is a super common pitfall when working with sequential operations and error handling in RxJava. Let's walk through exactly how to make your download sequence skip failed tasks and keep going to the next URL.

The Problem with Your Current Setup #

Right now, when downloadObservable(url) throws an error, it propagates up to the concatMap operator, which by default terminates the entire observable sequence. Using onErrorResumeNext() directly on the outer observable would replace the rest of the sequence entirely, not just skip the failed single task—so that's why it isn't working for your use case.

The Solution: Handle Errors Per-Download #

The key is to contain the error within each individual download observable, so it doesn't break the entire sequence. We can do this by adding onErrorResumeNext() inside the concatMap lambda, for each downloadObservable(url) call. This way, when a download fails, we just signal that this task is done (without terminating the whole stream) and move on to the next URL.

Example Code #

First, here's a simple version that skips failed downloads and logs the error:

Observable.from(urls)
    .concatMap(url -> 
        downloadObservable(url)
            // Catch errors for THIS specific download only
            .onErrorResumeNext(throwable -> {
                // Log the failure for debugging
                System.err.printf("Failed to download %s: %s%n", url, throwable.getMessage());
                // Return an empty observable to signal this task is "completed"
                return Observable.empty();
            })
    )
    .subscribe(
        successfulResult -> {
            // Handle your successful download result here
            System.out.printf("Successfully downloaded: %s%n", successfulResult);
        },
        unexpectedError -> {
            // This will only trigger if the source (Observable.from(urls)) has an error
            System.err.printf("Unexpected sequence error: %s%n", unexpectedError.getMessage());
        },
        () -> {
            System.out.println("All download tasks finished (skipped failures)");
        }
    );

If You Need to Track Failures #

If you want to keep track of both successful and failed downloads (instead of just skipping failures), create a simple wrapper class to hold the status of each task:

// Simple class to hold download status
class DownloadResult {
    private final String url;
    private final boolean isSuccess;
    private final Throwable error;

    public DownloadResult(String url, boolean isSuccess, Throwable error) {
        this.url = url;
        this.isSuccess = isSuccess;
        this.error = error;
    }

    // Getters for url, isSuccess, error
}

Then modify the observable chain to emit these results:

Observable.from(urls)
    .concatMap(url -> 
        downloadObservable(url)
            // Map successful download to a success result
            .map(result -> new DownloadResult(url, true, null))
            // Catch errors and emit a failure result
            .onErrorResumeNext(throwable -> 
                Observable.just(new DownloadResult(url, false, throwable))
            )
    )
    .subscribe(result -> {
        if (result.isSuccess()) {
            System.out.printf("Success: %s%n", result.getUrl());
        } else {
            System.err.printf("Failure: %s - %s%n", result.getUrl(), result.getError().getMessage());
        }
    },
    unexpectedError -> {
        System.err.printf("Sequence error: %s%n", unexpectedError.getMessage());
    },
    () -> {
        System.out.println("All download tasks completed");
    }
);

Why This Works #

By handling the error inside the concatMap lambda, we're telling RxJava: "If this specific download fails, don't kill the whole sequence—just treat this task as done and move to the next one." The outer observable never sees the error, so it keeps processing the next URL in your list.

内容的提问来源于stack exchange，提问作者krupal.agile