Spring Reactive递归调用API：基于响应参数获取全量Student方案问询

首先定义外部API的响应DTO，用来映射返回结果：

import lombok.Data;
import java.util.List;

@Data
public class StudentPageResponse {
    private List<Student> students;
    private int count; // 数据库剩余学生数
    private int offset; // 当前数据的偏移量
}

@Data
public class Student {
    private Long id;
    private String name;
    // 根据实际需求添加其他字段
}

接下来编写基于WebClient的外部API调用方法（非阻塞）：

import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

@Service
public class StudentService {
    private final WebClient webClient;

    public StudentService(WebClient.Builder webClientBuilder) {
        // 替换为实际的外部API地址
        this.webClient = webClientBuilder.baseUrl("https://your-external-api.com").build();
    }

    // 分页获取学生列表，offset为起始偏移量，pageSize为每页数量
    public Mono<StudentPageResponse> fetchStudents(int offset, int pageSize) {
        return webClient.get()
                .uri(uriBuilder -> uriBuilder
                        .path("/students")
                        .queryParam("offset", offset)
                        .queryParam("limit", pageSize)
                        .build())
                .retrieve()
                .bodyToMono(StudentPageResponse.class)
                .onErrorResume(e -> {
                    // 可根据需求添加异常处理逻辑
                    return Mono.error(new RuntimeException("调用外部API失败", e));
                });
    }
}

核心实现：用expand操作符递归分页获取所有学生 #

expand是Spring Reactive中专门用于递归展开异步序列的操作符，完全符合非阻塞的Reactive理念，比Flux.generate更适合这个场景：

import reactor.core.publisher.Flux;

public Flux<Student> getAllStudents() {
    final int PAGE_SIZE = 2; // 每页最多2个学生

    // 初始调用：从offset=0开始获取第一页
    return fetchStudents(0, PAGE_SIZE)
            // 递归展开：对每个响应，判断是否需要继续获取下一页
            .expand(response -> {
                int nextOffset = response.getOffset() + PAGE_SIZE;
                // 当剩余学生数count>0时，继续调用下一页；否则返回Mono.empty()终止递归
                return response.getCount() > 0 ? fetchStudents(nextOffset, PAGE_SIZE) : Mono.empty();
            })
            // 将每个分页响应中的学生列表转换为Flux<Student>
            .flatMapIterable(StudentPageResponse::getStudents);
}

为什么不用Flux.generate？ #

Flux.generate是基于同步状态生成序列的操作符，若在其中调用异步的API方法，需要阻塞等待结果（比如用block()），这会破坏Reactive的非阻塞特性。而expand天然支持异步递归，是更合适的选择。

额外说明 #

• 如果外部API的count字段不可靠，可以改为判断返回的students列表是否为空来终止递归：

  .expand(response -> {
      int nextOffset = response.getOffset() + PAGE_SIZE;
      return !response.getStudents().isEmpty() ? fetchStudents(nextOffset, PAGE_SIZE) : Mono.empty();
  })
  

• 可以根据需求添加超时、重试等增强逻辑，比如在fetchStudents方法中添加.timeout(Duration.ofSeconds(10))或.retry(3)。

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