映入 webflux 依赖
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId></dependency>
ObsFluxUtil 工具类
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.codec.multipart.FilePart;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;@Component
public class ObsFluxUtil {/*** 上传文件（文件id系统自动生成）*/public Mono<Map<String, Object>> uploadFile(FilePart filePart) {// 复用到私有通用方法return doUpload(filePart, null);}/*** 上传文件，支持自定义文件id（objectKey）*/public Mono<Map<String, Object>> uploadFile(FilePart filePart, String objectKey) {return doUpload(filePart, objectKey);}/*** 通用上传实现*/private Mono<Map<String, Object>> doUpload(FilePart filePart, String specifiedObjectKey) {String originalFilename = filePart.filename();String suffix = "";if (originalFilename.contains(".")) {suffix = originalFilename.substring(originalFilename.lastIndexOf("."));}String objectKey = (specifiedObjectKey == null || specifiedObjectKey.trim().isEmpty())? UUID.randomUUID().toString().replace("-", ""): specifiedObjectKey;Map<String, Object> resultMap = new HashMap<>();resultMap.put("objectKey", objectKey);resultMap.put("originalFilename", originalFilename);resultMap.put("suffix", suffix);return DataBufferUtils.join(filePart.content()).flatMap(dataBuffer -> {byte[] bytes = new byte[dataBuffer.readableByteCount()];dataBuffer.read(bytes);DataBufferUtils.release(dataBuffer);// 在 boundedElastic 线程池中执行阻塞上传逻辑return Mono.fromCallable(() -> {try (InputStream inputStream = new ByteArrayInputStream(bytes)) {boolean success = ObsUtil.uploadStream(objectKey, inputStream);resultMap.put("success", success);return resultMap;}}).subscribeOn(Schedulers.boundedElastic());});}/*** 从华为云 OBS 下载文件*/public Mono<Void> downloadFile(String objectKey, ServerHttpResponse response) {return Mono.fromCallable(() -> ObsUtil.downloadStream(objectKey)).subscribeOn(Schedulers.boundedElastic()).flatMap(inputStream -> {if (inputStream == null) {response.setStatusCode(HttpStatus.NOT_FOUND);return response.setComplete();}response.getHeaders().set(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=" + objectKey);response.getHeaders().setContentType(MediaType.APPLICATION_OCTET_STREAM);Flux<DataBuffer> dataBufferFlux = DataBufferUtils.readInputStream(() -> inputStream,new DefaultDataBufferFactory(),4096);return response.writeWith(dataBufferFlux).publishOn(Schedulers.boundedElastic()).doFinally(signalType -> {try {inputStream.close();} catch (Exception ignored) {}});});}/*** 图片预览*/public Mono<Void> previewImage(String objectKey, ServerHttpResponse response) {return Mono.fromCallable(() -> ObsUtil.downloadStream(objectKey)).subscribeOn(Schedulers.boundedElastic()).flatMap(inputStream -> {if (inputStream == null) {response.setStatusCode(HttpStatus.NOT_FOUND);return response.setComplete();}String lowerKey = objectKey.toLowerCase();MediaType mediaType = MediaType.IMAGE_JPEG;if (lowerKey.endsWith(".png")) {mediaType = MediaType.IMAGE_PNG;} else if (lowerKey.endsWith(".gif")) {mediaType = MediaType.IMAGE_GIF;}response.getHeaders().setContentType(mediaType);Flux<DataBuffer> dataBufferFlux = DataBufferUtils.readInputStream(() -> inputStream,new DefaultDataBufferFactory(),4096);return response.writeWith(dataBufferFlux).publishOn(Schedulers.boundedElastic()).doFinally(signalType -> {try {inputStream.close();} catch (Exception ignored) {}});});}// ====================== 批量上传（复用 uploadFile） ======================/*** 批量上传文件（默认使用单文件 uploadFile 的 objectKey 生成逻辑）*/public Mono<List<Map<String, Object>>> uploadFiles(List<FilePart> fileParts) {int defaultConcurrency = Math.max(1, Math.min(4, Runtime.getRuntime().availableProcessors()));return uploadFiles(fileParts, null, defaultConcurrency);}/*** 批量上传文件（支持自定义 objectKey 生成规则和并发度；内部复用 uploadFile）* @param fileParts 文件列表* @param objectKeyGenerator 自定义 objectKey 生成器（参数：originalFilename, index），为 null 则使用 uploadFile 默认规则* @param concurrency 并发度（建议不超过 CPU 核心数的 2 倍）*/public Mono<List<Map<String, Object>>> uploadFiles(List<FilePart> fileParts,BiFunction<String, Integer, String> objectKeyGenerator,int concurrency) {AtomicInteger idx = new AtomicInteger(0);int conc = Math.max(1, concurrency);return Flux.fromIterable(fileParts).flatMap(filePart -> {int i = idx.getAndIncrement();String originalFilename = filePart.filename();// 根据是否提供生成器，决定调用哪个 uploadFile 重载Mono<Map<String, Object>> singleUploadMono;if (objectKeyGenerator != null) {String objectKey = objectKeyGenerator.apply(originalFilename, i);singleUploadMono = uploadFile(filePart, objectKey);} else {singleUploadMono = uploadFile(filePart);}// 统一错误兜底，保证批量上传不中断return singleUploadMono.onErrorResume(ex -> {Map<String, Object> fallback = new HashMap<>();fallback.put("originalFilename", originalFilename);if (objectKeyGenerator != null) {// 尽力放入生成过的 key，便于排查try {fallback.put("objectKey", objectKeyGenerator.apply(originalFilename, i));} catch (Exception ignored) { }}fallback.put("success", false);fallback.put("error", ex.getMessage());return Mono.just(fallback);});}, conc).collectList();}}
Controller 测试
import com.example.demo.utils.ObsFluxUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.codec.multipart.FilePart;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;import java.util.List;
import java.util.Map;/*** @author chenlong* @create 2025-05-16 17:14*/
@Slf4j
@RestController
@RequestMapping("/obs")
public class DemoController {@Autowiredprivate ObsFluxUtil obsFluxUtil;@PostMapping(value = "/uploadFile")public Mono<Map<String, Object>> uploadFile(@RequestPart(value = "file") FilePart filePart) {return obsFluxUtil.uploadFile(filePart);}@PostMapping(value = "/uploadFiles")public Mono<List<Map<String, Object>>> uploadFiles(@RequestPart("files") List<FilePart> fileParts) {return obsFluxUtil.uploadFiles(fileParts);}@GetMapping(value = "/previewImage")public Mono<Void> previewImage(@RequestParam(value = "objectKey") String objectKey, ServerHttpResponse response) {return obsFluxUtil.previewImage(objectKey, response);}@GetMapping("/downloadFile")public Mono<Void> downloadFile(@RequestParam String objectKey, ServerHttpResponse response) {return obsFluxUtil.downloadFile(objectKey, response);}}