如何用WebClient循环调用REST API并以非阻塞方式处理响应

问题场景 #

需要调用最多支持7天日期范围的getLoanDetails REST API，处理任意时长的输入日期范围（如30天、半年），需拆分多次调用API。目标是解析API返回的JSON生成Person对象，并收集所有贷款额最小的Person。当前已完成日期范围拆分，但不知道如何非阻塞处理所有API响应，避免使用block()方法。

解决方案 #

用Reactor响应式操作符替代同步循环，异步发起所有API调用，合并结果后计算全局最小贷款额并过滤出对应Person：

• 将拆分后的日期范围列表转为Flux，实现异步批量调用API
• 每个日期段调用API后，解析JSON生成List<Person>并展开为Flux<Person>
• 收集所有Person到列表，计算全局最小贷款额
• 过滤出所有贷款额等于最小值的Person，返回结果

完整代码实现 #

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.http.MediaType;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.util.List;
import java.util.Comparator;

// 自定义日期范围记录类，替代原Pair<LocalDate>
record DateRange(LocalDate startDate, LocalDate endDate) {}

record Person(String name, BigDecimal loanAmount, LocalDate dateOfBirth) {}

public class LoanService {
    private final String apiURI;
    private final String apiKey;

    public LoanService(String apiURI, String apiKey) {
        this.apiURI = apiURI;
        this.apiKey = apiKey;
    }

    Flux<Person> findPeopleWithMinimumLoan(LocalDate startDate, LocalDate endDate) {
        List<DateRange> dateRanges = getDates(startDate, endDate);

        return Flux.fromIterable(dateRanges)
                // 异步发起每个日期段的API调用，展开结果为Person流
                .flatMap(this::fetchAndParsePersons)
                // 收集所有Person到内存列表
                .collectList()
                // 计算全局最小贷款额并过滤对应Person
                .flatMapMany(persons -> {
                    if (persons.isEmpty()) {
                        return Flux.empty();
                    }
                    // 找出所有Person中的最小贷款额
                    BigDecimal minLoanAmount = persons.stream()
                            .map(Person::loanAmount)
                            .min(Comparator.naturalOrder())
                            .orElse(BigDecimal.ZERO);
                    // 过滤出贷款额等于最小值的Person
                    return Flux.fromIterable(persons)
                            .filter(person -> person.loanAmount().compareTo(minLoanAmount) == 0);
                });
    }

    // 封装API调用与JSON解析逻辑
    private Mono<List<Person>> fetchAndParsePersons(DateRange dateRange) {
        return callLoanApi(dateRange.startDate(), dateRange.endDate())
                .map(this::findPeopleWithMinimumLoanAmount);
    }

    Mono<String> callLoanApi(LocalDate startDate, LocalDate endDate) {
        return WebClient.create(apiURI)
                .get()
                .uri(uriBuilder -> uriBuilder
                        .queryParam("aoi_key", apiKey)
                        .queryParam("start_date", startDate)
                        .queryParam("end_date", endDate)
                        .build())
                .accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .bodyToMono(String.class);
    }

    // 已实现的JSON转List<Person>解析方法
    List<Person> findPeopleWithMinimumLoanAmount(String jsonFromLoanApi) {
        // 此处为JSON解析逻辑，根据实际API返回格式实现
        return List.of();
    }

    // 已实现的日期范围拆分方法
    List<DateRange> getDates(LocalDate startDate, LocalDate endDate) {
        // 此处为日期拆分逻辑，将长范围拆分为多个7天内的分段
        return List.of();
    }
}

关键说明 #

• 异步并发调用：用Flux.fromIterable替代同步for循环，让Reactor自动调度API调用的并发执行，提升效率
• 全程非阻塞：所有操作基于Reactor的Mono/Flux实现，无block()调用，符合响应式编程规范
• 全局最小值计算：在收集所有结果后一次性计算最小贷款额，避免流式计算中重复遍历
• 空值防护：增加空列表判断，避免空指针异常

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