Spring Webflux WebClient如何返回subscribe()的执行结果？——单核心环境下兼顾请求并发与结果校验的方案咨询

Great question! Let's break down what's going wrong with your current approaches and fix this with a solution that balances concurrency, result collection, and proper program lifecycle.

What's Wrong With Your Current Implementations #

• First approach (using block()): Calling block() on each request Mono forces sequential execution—each request blocks the single thread until it finishes, so 1000 requests take 10,000 seconds total. That's why you only see one log every 10 seconds.
• Second approach (fire-and-forget subscribe()): subscribe() runs asynchronously, but your main thread doesn't wait for these async tasks to complete. As soon as the main method exits, the JVM terminates, killing in-flight requests before they can log results or return responses.
• Third approach (callback in subscribe()): While this enables concurrent requests, the callback logic is isolated from your method's return value. The main method returns immediately, and you can't propagate the condition check result back to the caller.

The Middle Ground: Concurrent Execution + Wait for All Results + Result Handling #

Since you're on a single-core environment, you don't need a thread pool—Reactor's non-blocking model can handle hundreds of concurrent HTTP requests efficiently on one thread. Here's how to implement the solution:

1. Wrap all requests in a Flux: Convert your collection of someRequests into a reactive stream.
2. Use flatMap for concurrency: This operator subscribes to multiple request Monos concurrently, sending all requests to the server quickly.
3. Handle per-response logic: Use doOnNext to log condition checks as each response arrives.
4. Collect results and wait: Use collectList() to aggregate all responses into a single Mono<List<String>>, then block() to wait for all requests to complete before proceeding.
5. Final condition check: Once you have all responses, evaluate your condition (e.g., check if any/all meet criteria) and return the appropriate result.

Example Code

// Assume allSomeRequests is your collection of 1000 request objects
Flux<String> responseFlux = Flux.fromIterable(allSomeRequests)
    // flatMap with concurrency limit (adjust based on the server's capacity)
    .flatMap(request -> 
        webClient.post()
            .uri("https://api-that-will-take-ten-seconds.com/slow")
            .body(BodyInserters.fromValue(request))
            .retrieve()
            .bodyToMono(String.class)
            // Log condition check as each response arrives
            .doOnNext(response -> {
                if (someCondition(response)) {
                    System.out.println("Condition met for request");
                } else {
                    System.out.println("Condition not met for request");
                }
            })
            // Handle errors to prevent one failed request from breaking the whole stream
            .onErrorResume(throwable -> {
                System.err.println("Request failed: " + throwable.getMessage());
                return Mono.empty(); // Or return a default value if needed
            })
    , 100); // Concurrency limit—100 works well for single-core; tweak based on server tolerance

// Wait for all requests to complete and collect all responses
List<String> allResponses = responseFlux.collectList().block();

// Evaluate your final condition (adjust logic to match your needs)
boolean anyConditionMet = allResponses.stream().anyMatch(this::someCondition);
if (anyConditionMet) {
    System.out.println("At least one request met the condition");
    return "success";
} else {
    System.out.println("No requests met the condition");
    return "bad";
}

Key Details #

• Concurrency with flatMap: The second parameter sets the maximum number of concurrent requests. On a single core, you can safely set this to a high value (like 100 or 200) because Reactor uses non-blocking I/O—your thread won't be stuck waiting for responses, it'll handle other tasks while waiting.
• Waiting with block(): Calling block() on the collected Mono<List<String>> ensures your main thread waits until all requests complete (successfully or with errors handled) before proceeding. This prevents the JVM from exiting early.
• Error Handling: onErrorResume ensures a single failed request doesn't crash the entire stream. Customize it to log errors and return fallback values if needed.

If you don't need to collect all responses (e.g., you only care if any response meets the condition), optimize with the any() operator:

boolean conditionMet = responseFlux.any(this::someCondition).block();

This short-circuits as soon as it finds a matching response, saving you from waiting for all requests to finish.

内容的提问来源于stack exchange，提问作者PatPanda