MongoDB异步FindIterable迭代如何在WebFlux场景下取消？

你遇到的这个问题很典型——当用WebFlux流式传输大量MongoDB结果时，客户端断开连接后Flowable虽然收到了取消信号，但Mongo的异步迭代还在后台跑，浪费资源对吧？好在针对org.mongodb:mongodb-driver-async:3.6.3版本，我们可以通过手动绑定MongoCursor的生命周期和Flowable的订阅来解决这个问题。

核心思路是：在创建Flowable时，明确监听取消事件，一旦收到取消信号就立即关闭Mongo的游标，终止异步迭代。下面是具体的实现方案：

实现代码示例 #

import io.reactivex.Flowable;
import io.reactivex.disposables.Disposables;
import org.mongodb.async.client.MongoCursor;
import org.mongodb.async.client.FindIterable;
import com.mongodb.async.SingleResultCallback;

import java.io.IOException;
import java.util.function.Consumer;

private static <T> Flowable<T> toFlowable(FindIterable<T> findIterable) {
    return Flowable.create(emitter -> {
        // 获取Mongo异步游标
        MongoCursor<T> cursor = findIterable.cursor();

        // 绑定取消事件：当Flowable被取消时，关闭游标终止迭代
        emitter.setDisposable(Disposables.fromRunnable(() -> {
            try {
                cursor.close();
            } catch (IOException e) {
                // 处理关闭时的异常，可根据需求选择抛出或记录日志
                if (!emitter.isCancelled()) {
                    emitter.onError(e);
                }
            }
        }));

        // 定义错误处理逻辑：出错时关闭游标并通知Flowable
        Consumer<Throwable> errorHandler = err -> {
            if (!emitter.isCancelled()) {
                emitter.onError(err);
            }
            try {
                cursor.close();
            } catch (IOException e) {
                // 日志记录即可，避免二次抛出
                // log.error("Failed to close cursor after error", e);
            }
        };

        // 递归异步获取下一条数据的回调
        SingleResultCallback<T> resultCallback = (result, err) -> {
            if (err != null) {
                errorHandler.accept(err);
                return;
            }

            if (result != null) {
                if (!emitter.isCancelled()) {
                    emitter.onNext(result);
                    // 继续获取下一条数据
                    cursor.tryNext(resultCallback);
                } else {
                    // 已取消，直接关闭游标
                    try {
                        cursor.close();
                    } catch (IOException e) {
                        errorHandler.accept(e);
                    }
                }
            } else {
                // 没有更多数据了，完成流并关闭游标
                if (!emitter.isCancelled()) {
                    emitter.onComplete();
                }
                try {
                    cursor.close();
                } catch (IOException e) {
                    errorHandler.accept(e);
                }
            }
        };

        // 启动异步迭代
        cursor.tryNext(resultCallback);
    }, BackpressureStrategy.BUFFER); // 根据业务场景选择合适的背压策略
}

关键细节说明 #

• 游标生命周期绑定：通过emitter.setDisposable()注册一个取消回调，当Flowable被取消（比如客户端断开连接）时，立即调用cursor.close()终止Mongo的异步迭代，释放服务器端和客户端的资源。
• 异步迭代的递归处理：利用Mongo异步游标tryNext()方法的回调特性，递归获取下一条数据，同时在每一步检查Flowable是否已取消，避免无效的后续请求。
• 错误与资源清理：无论正常完成、出错还是取消，都会确保游标被关闭，避免资源泄漏。
• 背压策略选择：示例中用了BackpressureStrategy.BUFFER，你可以根据业务需求调整——比如如果不需要保留所有未处理的数据，可选择DROP或LATEST。

验证效果 #

当客户端断开连接时，WebFlux会触发Flowable的取消信号，此时我们注册的Disposable会执行cursor.close()，Mongo驱动会立即停止从服务器拉取数据，并释放相关的游标资源，不会再在后台继续迭代。

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