WebFlux函数式端点：Flux返回值的异常处理与HTTP状态码返回

Great question! I’ve run into this exact issue a few times when working with WebFlux functional endpoints. The key difference between Mono and Flux here is that Flux is a sequence of elements—exceptions thrown in the stream can either terminate the entire flow or be handled per-element, depending on where you apply your error logic. Let’s break down the solutions based on what you’re trying to achieve:

1. Catch Flux exceptions to return an HTTP error status code (full failure) #

If you want the entire request to fail with an HTTP error (e.g., 400 Bad Request) when any element in the Flux throws an exception, the most reliable approach is to convert the Flux to a Mono first (like collecting it into a list), then apply onErrorResume to that Mono. This works because any exception in the Flux will propagate to the downstream Mono, where you can map it to the correct HTTP response.

Example Code: #

// Service layer returning a Flux that might throw exceptions
@Service
public class DataService {
    public Flux<Data> fetchDataFlux() {
        return Flux.just(1, 2, 3)
                .map(id -> {
                    if (id == 2) throw new BadRequestException("Invalid ID: " + id);
                    return new Data(id, "Data for " + id);
                });
    }
}

// Functional endpoint configuration
@Configuration
public class DataRouter {
    @Bean
    public RouterFunction<ServerResponse> dataRouter(DataService service) {
        return RouterFunctions.route(RequestPredicates.GET("/api/data"), request ->
                // Convert Flux to Mono via collectList()
                service.fetchDataFlux()
                        .collectList()
                        // On success, return 200 with the list
                        .flatMap(dataList -> ServerResponse.ok().bodyValue(dataList))
                        // Catch specific exceptions and map to HTTP status codes
                        .onErrorResume(BadRequestException.class, ex ->
                                ServerResponse.badRequest().bodyValue("Request failed: " + ex.getMessage()))
                        .onErrorResume(NotFoundException.class, ex ->
                                ServerResponse.notFound().build())
                        // Catch all other exceptions for a generic 500
                        .onErrorResume(ex ->
                                ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).bodyValue("Unexpected error"))
        );
    }
}

2. Skip failed elements and return partial data #

If you want to continue processing other elements when one fails (instead of failing the entire request), use Flux-specific error operators like onErrorContinue or onErrorReturn:

Using onErrorContinue (skip the failed element): #

This operator lets you log the error and continue processing the rest of the stream:

@Bean
public RouterFunction<ServerResponse> dataRouter(DataService service) {
    return RouterFunctions.route(RequestPredicates.GET("/api/data"), request ->
            service.fetchDataFlux()
                    // Skip the failed element and log the error
                    .onErrorContinue((ex, failedElement) -> {
                        log.warn("Failed to process element {}: {}", failedElement, ex.getMessage());
                    })
                    .collectList()
                    .flatMap(dataList -> ServerResponse.ok().bodyValue(dataList))
                    // Handle fatal errors that break the entire stream (e.g., DB connection failure)
                    .onErrorResume(ConnectException.class, ex ->
                            ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).bodyValue("Service unavailable"))
    );
}

Using onErrorReturn (replace the failed element with a default): #

If you want to substitute the failed element with a fallback value instead of skipping it:

@Bean
public RouterFunction<ServerResponse> dataRouter(DataService service) {
    return RouterFunctions.route(RequestPredicates.GET("/api/data"), request ->
            service.fetchDataFlux()
                    // Replace failed elements with a default Data object
                    .onErrorReturn(BadRequestException.class, new Data(-1, "Failed to load data"))
                    .collectList()
                    .flatMap(dataList -> ServerResponse.ok().bodyValue(dataList))
    );
}

Why your initial attempt might have failed #

Chances are you tried applying onErrorResume directly to the Flux but didn’t account for how exceptions propagate:

• If you added onErrorResume to the Flux but then converted it to a Mono (e.g., with collectList), the exception might have been swallowed if you returned an empty Flux from onErrorResume—instead, you need to let the exception propagate to the Mono stage to trigger an HTTP error.
• If you tried returning a Mono<ServerResponse> from the Flux’s onErrorResume, you’d get a type mismatch (since Flux expects a Flux<Data>, not a Mono<ServerResponse>).

By converting the Flux to a Mono first, you align the error handling pattern with what you already know works for Mono results.

内容的提问来源于stack exchange，提问作者Nicolas Barbé