当前位置：首页 > article >正文

Reactor编程模型中,阻塞上传文件FilePart的3中方式

article 2025/1/18 23:17:32

1. 使用非阻塞的方式处理数据

将 Flux 转换为 Mono，并确保整个链式调用是非阻塞的。你可以使用 Mono.fromCallable 或其他非阻塞的方式来处理资源转换。

public Mono<Void> addDocumentsByFile(FilePart file) {
    String suffix = FileNameUtil.getSuffix(file.filename());
    
    return file.content()
        .reduce(DataBuffer::write)
        .map(DataBuffer::toByteBuffer)
        .map(ByteBuffer::array)
        .map(ByteArrayResource::new)
        .flatMap(resource -> {
            TikaDocumentReader tikaDocumentReader = new TikaDocumentReader(resource);
            List<Document> splitDocuments;
            if ("csv".equalsIgnoreCase(suffix)) {
                splitDocuments = new CsvTextSplitter().apply(tikaDocumentReader.read());
            } else {
                splitDocuments = new TokenTextSplitter().apply(tikaDocumentReader.read());
            }
            // 将文档存入向量数据库
            return Mono.fromRunnable(() -> {
                for (Document doc : splitDocuments) {
                    vectorStore.add(doc); // 假设vectorStore是一个异步接口
                }
            });
        })
        .then();
}

2. 使用 publishOn 切换到阻塞线程池

如果你确实需要使用阻塞操作，可以使用 publishOn 将操作切换到一个允许阻塞的线程池（例如 (Schedulers.boundedElastic())）。这会确保阻塞操作不会影响到非阻塞线程。

public Mono<Void> addDocumentsByFile(FilePart file) {
    String suffix = FileNameUtil.getSuffix(file.filename());

    return file.content()
        .reduce(DataBuffer::write)
        .map(DataBuffer::toByteBuffer)
        .map(ByteBuffer::array)
        .map(ByteArrayResource::new)
        .publishOn(Schedulers.boundedElastic())
        .flatMap(resource -> {
            TikaDocumentReader tikaDocumentReader = new TikaDocumentReader(resource);
            List<Document> splitDocuments;
            if ("csv".equalsIgnoreCase(suffix)) {
                splitDocuments = new CsvTextSplitter().apply(tikaDocumentReader.read());
            } else {
                splitDocuments = new TokenTextSplitter().apply(tikaDocumentReader.read());
            }
            // 将文档存入向量数据库
            for (Document doc : splitDocuments) {
                vectorStore.add(doc); // 假设vectorStore是一个同步接口
            }
            return Mono.empty();
        });
}

3. 使用 Mono.fromCallable 和 subscribeOn

如果你需要在一个特定的线程上执行阻塞操作，可以使用 Mono.fromCallable 并结合 subscribeOn 来确保操作在正确的线程上执行。

public Mono<Void> addDocumentsByFile(FilePart file) {
    String suffix = FileNameUtil.getSuffix(file.filename());

    return file.content()
        .reduce(DataBuffer::write)
        .map(DataBuffer::toByteBuffer)
        .map(ByteBuffer::array)
        .map(ByteArrayResource::new)
        .flatMap(resource -> Mono.fromCallable(() -> {
            TikaDocumentReader tikaDocumentReader = new TikaDocumentReader(resource);
            List<Document> splitDocuments;
            if ("csv".equalsIgnoreCase(suffix)) {
                splitDocuments = new CsvTextSplitter().apply(tikaDocumentReader.read());
            } else {
                splitDocuments = new TokenTextSplitter().apply(tikaDocumentReader.read());
            }
            // 将文档存入向量数据库
            for (Document doc : splitDocuments) {
                vectorStore.add(doc); // 假设vectorStore是一个同步接口
            }
            return null;
        }).subscribeOn(Schedulers.boundedElastic()))
        .then();
}