背景最近在阅读Spring 6.x源码时发现框架内部有不少地方使用了StreamSupport和Spliterator来处理各种数据源。研究了一下后突然意识到这套机制不仅能处理集合还能定制任何形式的数据流。那么我就可以尝试用它来分页读取第三方的数据了。正好项目里有个定时任务需要从第三方平台拉取商品数据每天同步一次。原来的代码用的是do-while循环分页请求每页100条具体的代码如下/** * 商品实体 */ Data public class Product { private Long id; private String name; } /** * 拉取第三方商品数据 */ public void syncProductData() { int currentPage 1; boolean hasMoreData true; int pageSize 100; do { // 调用第三方API获取一页数据 JSONObject requestBody new JSONObject(); requestBody.put(page, currentPage); requestBody.put(pageSize, pageSize); JSONArray productArray requestThirdPartyApi(requestBody); // 判断是否还有下一页 if (productArray.size() pageSize) { hasMoreData false; } // 转换并保存数据 ListProduct products parseProducts(productArray); saveProducts(products); currentPage; log.info(已同步第 {} 页本页 {} 条记录, currentPage, products.size()); } while (hasMoreData); log.info(商品数据同步完成); } private JSONArray requestThirdPartyApi(JSONObject body) { // 模拟调用第三方API String response httpClient.post(/api/products, body); JSONObject result JSON.parseObject(response); return result.getJSONArray(data); } private ListProduct parseProducts(JSONArray array) { return array.stream() .map(obj - { JSONObject json (JSONObject) obj; Product product new Product(); product.setId(json.getLong(id)); product.setName(json.getString(name)); return product; }) .collect(Collectors.toList()); } private void saveProducts(ListProduct products) { // 批量保存到数据库 productRepository.batchInsert(products); }这代码能跑但有几个问题可变状态太多currentPage、hasMoreData 这些变量在循环里改来改去分页逻辑和业务逻辑耦合拉数据、判断是否有下一页、保存数据全挤在一起复用性差如果另一个接口也要分页拉取得再写一遍类似的循环能不能用 Java 17 的 Stream API 重构我们来试试看哈。核心思路Stream API 本身不支持分页场景但可以通过Spliterator定制数据源。关键在于自定义 Spliterator实现 tryAdvance() 方法每次调用时拉取一页数据用 StreamSupport 转成 Stream把 Spliterator 包装成惰性 StreamStream流式处理逐条或分批处理商品数据改造后的代码可以这样用createProductStream(100) .peek(product - log.info(处理商品: {}, product.getName())) .forEach(this::saveProduct);是不是简洁多了实现步骤第一步定义分页数据源先抽象一个函数式接口表示”给定页码返回该页数据”FunctionalInterface public interface PageDataFetcher { JSONArray fetchPage(int pageNumber); }这样做的好处是不管是调商品接口还是其他第三方系统只要实现这个接口就行。第二步实现自定义 Spliterator/** * 分页数据流迭代器 */ public class PagingSpliterator extends Spliterators.AbstractSpliteratorProduct { private final PageDataFetcher fetcher; private final int pageSize; private int currentPage 1; private IteratorProduct currentPageIterator Collections.emptyIterator(); private boolean hasMore true; public PagingSpliterator(PageDataFetcher fetcher, int pageSize) { super(Long.MAX_VALUE, Spliterator.ORDERED | Spliterator.NONNULL); this.fetcher fetcher; this.pageSize pageSize; } Override public boolean tryAdvance(Consumer? super Product action) { // 当前页还有数据直接返回 if (currentPageIterator.hasNext()) { action.accept(currentPageIterator.next()); return true; } // 当前页数据已消费完判断是否需要拉取下一页 if (!hasMore) { return false; } // 拉取下一页数据 JSONArray pageData fetcher.fetchPage(currentPage); if (pageData null || pageData.isEmpty()) { hasMore false; return false; } // 判断是否还有下一页 if (pageData.size() pageSize) { hasMore false; } // 将JSONArray转换为Product列表 ListProduct products parseProducts(pageData); currentPageIterator products.iterator(); currentPage; // 递归调用返回当前页的第一条数据 return tryAdvance(action); } private ListProduct parseProducts(JSONArray array) { return array.stream() .map(obj - { JSONObject json (JSONObject) obj; Product product new Product(); product.setId(json.getLong(id)); product.setName(json.getString(name)); return product; }) .collect(Collectors.toList()); } }关键点说明tryAdvance() 是 Spliterator 的核心方法每次 Stream 消费元素时都会调用currentPageIterator 缓存当前页的数据避免重复请求当前页消费完后自动拉取下一页懒加载通过 pageData.size() pageSize 判断是否还有下一页内部自动完成 JSON 到 Product 对象的转换第三步创建分页 Stream 的工厂方法/** * 创建商品分页数据流 * * param pageSize 每页大小 * return 商品流 */ private StreamProduct createProductStream(int pageSize) { // 定义分页数据获取逻辑 PageDataFetcher fetcher pageNumber - { JSONObject body new JSONObject(); body.put(page, pageNumber); body.put(pageSize, pageSize); return requestThirdPartyApi(body); }; // 创建 Spliterator PagingSpliterator spliterator new PagingSpliterator(fetcher, pageSize); // 转换成 Stream非并行 return StreamSupport.stream(spliterator, false); } private JSONArray requestThirdPartyApi(JSONObject body) { // 模拟调用第三方API String response httpClient.post(/api/products, body); JSONObject result JSON.parseObject(response); return result.getJSONArray(data); }这里用 Lambda 表达式实现了 PageDataFetcher接口封装了第三方API的调用逻辑。第四步改造业务方法原来的 syncProductData() 方法可以简化成这样/** * 同步商品数据Stream版本 */ public void syncProductData() { int pageSize 100; AtomicInteger processedCount new AtomicInteger(0); createProductStream(pageSize) .peek(product - { int count processedCount.incrementAndGet(); if (count % 100 0) { log.info(已处理 {} 条记录, count); } }) .forEach(this::saveProduct); log.info(数据同步完成共处理 {} 条记录, processedCount.get()); } private void saveProduct(Product product) { productRepository.insert(product); }注意原代码是按页批量保存每页调用一次保存这里改成了逐条保存。如果要保持批量保存的逻辑可以用 Stream的分组方法public void syncProductDataBatch() { int pageSize 100; AtomicInteger processedCount new AtomicInteger(0); createProductStream(pageSize) .collect(Collectors.groupingBy( product - processedCount.getAndIncrement() / pageSize )) .forEach((batchIndex, products) - { saveProducts(products); log.info(保存第 {} 批数据共 {} 条, batchIndex 1, products.size()); }); log.info(数据同步完成共处理 {} 条记录, processedCount.get()); } private void saveProducts(ListProduct products) { productRepository.batchInsert(products); }完整代码示例下面是改造后的完整类import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import lombok.Data; import lombok.extern.slf4j.Slf4j; import java.util.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.stream.Collectors; import java.util.stream.Stream; import java.util.stream.StreamSupport; /** * 第三方商品数据同步服务Stream API 版本 * * author linsongbin */ Slf4j public class ProductSyncService { /** * 同步商品数据按页批量保存 */ public void syncProductData() { int pageSize 100; AtomicInteger processedCount new AtomicInteger(0); createProductStream(pageSize) .collect(Collectors.groupingBy( product - processedCount.getAndIncrement() / pageSize )) .forEach((batchIndex, products) - { saveProducts(products); log.info(保存第 {} 批数据共 {} 条, batchIndex 1, products.size()); }); log.info(数据同步完成共处理 {} 条记录, processedCount.get()); } /** * 创建商品分页数据流 * * param pageSize 每页大小 * return 商品流 */ private StreamProduct createProductStream(int pageSize) { PageDataFetcher fetcher pageNumber - { JSONObject body new JSONObject(); body.put(page, pageNumber); body.put(pageSize, pageSize); return requestThirdPartyApi(body); }; PagingSpliterator spliterator new PagingSpliterator(fetcher, pageSize); return StreamSupport.stream(spliterator, false); } private JSONArray requestThirdPartyApi(JSONObject body) { // 模拟调用第三方API String response httpClient.post(/api/products, body); JSONObject result JSON.parseObject(response); return result.getJSONArray(data); } private void saveProducts(ListProduct products) { productRepository.batchInsert(products); } /** * 商品实体 */ Data public static class Product { private Long id; private String name; } /** * 分页数据获取器 */ FunctionalInterface interface PageDataFetcher { JSONArray fetchPage(int pageNumber); } /** * 分页数据流迭代器 */ static class PagingSpliterator extends Spliterators.AbstractSpliteratorProduct { private final PageDataFetcher fetcher; private final int pageSize; private int currentPage 1; private IteratorProduct currentPageIterator Collections.emptyIterator(); private boolean hasMore true; public PagingSpliterator(PageDataFetcher fetcher, int pageSize) { super(Long.MAX_VALUE, Spliterator.ORDERED | Spliterator.NONNULL); this.fetcher fetcher; this.pageSize pageSize; } Override public boolean tryAdvance(Consumer? super Product action) { if (currentPageIterator.hasNext()) { action.accept(currentPageIterator.next()); return true; } if (!hasMore) { return false; } JSONArray pageData fetcher.fetchPage(currentPage); if (pageData null || pageData.isEmpty()) { hasMore false; return false; } if (pageData.size() pageSize) { hasMore false; } ListProduct products parseProducts(pageData); currentPageIterator products.iterator(); currentPage; return tryAdvance(action); } private ListProduct parseProducts(JSONArray array) { return array.stream() .map(obj - { JSONObject json (JSONObject) obj; Product product new Product(); product.setId(json.getLong(id)); product.setName(json.getString(name)); return product; }) .collect(Collectors.toList()); } } }实际运行效果测试环境Java 17Windows 11拉取第三方平台商品数据共350条记录分4页控制台输出保存第 1 批数据共 100 条 保存第 2 批数据共 100 条 保存第 3 批数据共 100 条 保存第 4 批数据共 50 条 数据同步完成共处理 350 条记录对比原来的 do-while 版本日志输出更清晰代码行数从60行 减少到30行业务方法部分。技术细节解析Spliterator 的特性标记构造函数里有这么一行super(Long.MAX_VALUE, Spliterator.ORDERED | Spliterator.NONNULL);Long.MAX_VALUE预估大小未知分页数据无法提前知道总数ORDERED保证元素顺序按页码顺序拉取NONNULL元素不会为 null这些特性会影响 Stream 的优化策略。比如声明了ORDERED后续调用parallel()时会保留顺序。为什么用递归调用 tryAdvancereturn tryAdvance(action);这行代码看着有点奇怪为什么要递归因为拉取下一页后需要立即消费该页的第一条数据。如果直接 return true调用方还需要再次调用 tryAdvance() 才能拿到数据增加了复杂度。递归调用可以一次性完成”拉取消费”的逻辑。延迟加载的好处Stream 是惰性求值的只有在调用终止操作如forEach、collect时才会真正执行。这意味着数据不会一次性全部加载到内存可以随时中断流操作比如用 limit(10) 只拉取前10条如果用原来的 do-while必须等所有页拉取完才能处理。通用性扩展这套方案不限于商品接口任何分页 API 都能复用。比如对接订单数据Data public class Order { private Long id; private String orderNo; private BigDecimal amount; } // 创建订单分页流 StreamOrder orderStream createPagingStream( pageNumber - { JSONObject request new JSONObject(); request.put(page, pageNumber); request.put(pageSize, 100); return orderApiClient.queryOrders(request); }, 100 // 每页大小 ); orderStream .filter(order - order.getAmount().compareTo(new BigDecimal(1000)) 0) .forEach(this::saveOrder);核心就是提供一个PageDataFetcher实现即可。注意事项不要并行化StreamSupport.stream(spliterator, false)第二个参数必须是false因为分页请求有先后顺序异常处理如果fetchPage()抛异常Stream 会立即终止需要在 Spliterator 内部捕获日志打印建议在tryAdvance()里打日志方便排查问题总结这次改造的核心收获Stream API 不止能处理集合通过 Spliterator 可以处理任何数据源函数式编程让分页逻辑和业务逻辑彻底解耦惰性求值机制天然适合分页场景如果你的项目里也有类似的分页拉取逻辑不妨试试这个方案。代码会更优雅维护成本也更低。