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

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();
}

总结

为了避免阻塞操作导致的问题，建议尽量使用非阻塞的方式处理数据流。如果必须使用阻塞操作，可以通过 publishOn 或 subscribeOn 将操作切换到允许阻塞的线程池。这样可以确保你的应用程序在非阻塞环境中正常工作。