使用Project Reactor读取Cassandra全量记录时出现栈溢出问题

问题根源 #

你遇到的栈溢出，本质是递归式的异步分页调用导致栈帧累积：每次在fetchMore的回调里同步处理下一页数据并触发下一次fetchMore，当数据页数足够多时，线程栈会被不断叠加的调用帧撑爆。

核心解决方案：异步解耦分页逻辑 #

要避免栈溢出，必须打破同步递归的调用链，确保每一页的获取和处理在独立的栈帧中执行。结合Cassandra驱动的异步API和Reactor的特性，推荐用Flux.create配合异步回调+调度器的方式实现：

步骤1：正确使用Cassandra驱动的异步分页 #

Cassandra驱动3.6版本的Result类提供了fetchMoreAsync()方法，返回CompletionStage<Result>，用这个异步方法替代同步调用，避免阻塞线程。

步骤2：用Flux.create实现非阻塞流式输出 #

在Flux.create的回调中，先处理当前页的所有数据，再异步触发下一页的获取，并且通过调度器将下一页的处理切换到新的执行上下文，打破栈累积。

示例代码 #

// 简化的Cassandra Result类（模拟官方实现）
public class Result {
    private final List<String> data;
    private boolean hasMorePages;

    public Result(List<String> data, boolean hasMorePages) {
        this.data = data;
        this.hasMorePages = hasMorePages;
    }

    public int getAvailableWithoutFetching() {
        return data.size();
    }

    public List<String> allRows() {
        return data;
    }

    public CompletionStage<Result> fetchMoreAsync() {
        // 模拟异步获取下一页数据
        return CompletableFuture.supplyAsync(() -> {
            if (!hasMorePages) return null;
            hasMorePages = false; // 模拟最后一页
            return new Result(Arrays.asList("row-101", "row-102"), false);
        });
    }

    public boolean hasMorePages() {
        return hasMorePages;
    }
}

// 正确的全量读取实现
public Flux<String> fetchAllData(Result initialResult) {
    return Flux.create(sink -> {
        // 定义递归处理分页的方法
        Consumer<Result> processPage = result -> {
            // 发射当前页所有数据
            result.allRows().forEach(sink::next);

            if (result.hasMorePages()) {
                // 异步获取下一页，并用调度器切换上下文避免栈累积
                result.fetchMoreAsync()
                      .thenApply(nextResult -> {
                          // 异步处理下一页，这里用CommonPool确保在新线程执行
                          CompletableFuture.runAsync(() -> processPage.accept(nextResult));
                          return null;
                      })
                      .exceptionally(ex -> {
                          sink.error(ex);
                          return null;
                      });
            } else {
                sink.complete();
            }
        };

        // 启动初始页处理
        processPage.accept(initialResult);
    });
}

关键细节说明 #

• 异步解耦：通过CompletableFuture.runAsync()将下一页的处理放到新的线程执行，避免在同一个栈帧中累积递归调用。
• 错误处理：在fetchMoreAsync的exceptionally回调中捕获异常并传递给Flux的sink.error()，确保错误能被下游感知。
• 按需流式返回：Flux.create会根据下游的需求（背压）控制数据发射，符合你“基于需求流式返回”的设计目标。

为什么之前的方案没生效？ #

• Flux generate：它是同步生成数据的模式，不适合处理异步的分页获取，强行用的话会导致阻塞或者栈溢出。
• Sinks：如果没有正确处理异步调度，直接在回调中同步触发下一次分页，依然会导致栈帧累积，必须配合异步执行上下文才能解决。

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