在Quarkus应用中实现大文件尤其是视频文件的预览服务时，传统的阻塞式文件读取和传输方式会一次性将整个文件加载到内存中，当文件体积较大时极易引发内存溢出，同时也会让请求响应等待时间过长。响应式流采用异步非阻塞的处理模式，支持按片段逐步读取和传输文件内容，能够大幅降低内存占用，提升服务的并发处理能力。
响应式流在Quarkus中的核心支持
Quarkus内置了对响应式编程的支持，其中io.smallrye.mutiny.Multi是响应式流的核心抽象类型，它可以代表一个异步的、可能产生多个元素的流。在处理大文件时，我们可以将文件的读取过程封装为Multi流，每个元素对应文件的一个数据片段，这样就能实现边读边传的效果，不需要等待整个文件读取完成再返回响应。
大文件预览服务的实现步骤
1. 添加依赖
首先需要在项目的pom.xml中添加Quarkus的RESTEasy Reactive和响应式IO相关依赖，确保能够使用响应式流处理请求和文件操作：
<dependencies>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-resteasy-reactive</artifactId>
</dependency>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-resteasy-reactive-jackson</artifactId>
</dependency>
<dependency>
<groupId>io.smallrye.reactive</groupId>
<artifactId>smallrye-mutiny-vertx-core</artifactId>
</dependency>
</dependencies>
2. 编写响应式文件读取工具类
我们可以封装一个工具类，用于将本地文件转换为响应式流，按指定大小分片读取文件内容：
import io.smallrye.mutiny.Multi;
import io.vertx.mutiny.core.Vertx;
import io.vertx.mutiny.core.file.AsyncFile;
import io.vertx.mutiny.core.file.FileSystem;
import java.nio.ByteBuffer;
public class ReactiveFileReader {
private final Vertx vertx;
private final int chunkSize;
public ReactiveFileReader(Vertx vertx, int chunkSize) {
this.vertx = vertx;
this.chunkSize = chunkSize;
}
// 将文件转换为响应式流，每个元素是一个字节数组片段
public Multi<byte[]> readFileAsStream(String filePath) {
FileSystem fileSystem = vertx.fileSystem();
return fileSystem.open(filePath, new io.vertx.core.file.OpenOptions())
.onItem().transformToMulti(asyncFile -> {
// 按chunkSize大小分片读取文件
return Multi.createFrom().emitter(emitter -> {
readChunk(asyncFile, emitter, 0);
});
});
}
// 递归读取文件片段
private void readChunk(AsyncFile asyncFile, MultiEmitter<? super byte[]> emitter, long position) {
ByteBuffer buffer = ByteBuffer.allocate(chunkSize);
asyncFile.read(buffer, position, buffer)
.subscribe().with(
readResult -> {
buffer.flip();
byte[] chunk = new byte[buffer.remaining()];
buffer.get(chunk);
emitter.emit(chunk);
// 如果读取到的内容等于分片大小，说明还有后续内容，继续读取
if (chunk.length == chunkSize) {
readChunk(asyncFile, emitter, position + chunkSize);
} else {
// 文件读取完成，关闭文件并结束流
asyncFile.close();
emitter.complete();
}
},
error -> {
asyncFile.close();
emitter.fail(error);
}
);
}
}
3. 实现预览接口
接下来编写REST接口，接收文件路径参数，调用响应式读取工具返回流式响应，同时设置正确的响应头支持视频等文件的预览：
import io.smallrye.mutiny.Multi;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.PathParam;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import org.jboss.resteasy.reactive.RestStreamElementType;
import io.vertx.mutiny.core.Vertx;
import jakarta.inject.Inject;
@Path("/preview")
public class FilePreviewResource {
@Inject
Vertx vertx;
// 分片大小设置为64KB，可根据实际场景调整
private static final int CHUNK_SIZE = 64 * 1024;
@GET
@Path("/{fileName}")
@Produces(MediaType.APPLICATION_OCTET_STREAM)
@RestStreamElementType(MediaType.APPLICATION_OCTET_STREAM)
public Response previewFile(@PathParam("fileName") String fileName) {
// 实际场景中需要校验文件路径安全性，避免路径遍历攻击
String filePath = "/data/files/" + fileName;
ReactiveFileReader reader = new ReactiveFileReader(vertx, CHUNK_SIZE);
Multi<byte[]> fileStream = reader.readFileAsStream(filePath);
// 设置响应头，支持视频等文件的预览，同时允许分段请求
return Response.ok(fileStream)
.header("Content-Type", getContentType(fileName))
.header("Accept-Ranges", "bytes")
.build();
}
// 根据文件名后缀返回对应的Content-Type
private String getContentType(String fileName) {
if (fileName.endsWith(".mp4")) {
return "video/mp4";
} else if (fileName.endsWith(".avi")) {
return "video/x-msvideo";
} else if (fileName.endsWith(".mov")) {
return "video/quicktime";
}
return MediaType.APPLICATION_OCTET_STREAM;
}
}
优化与注意事项
- 文件路径需要做严格的安全校验，避免用户传入
../等路径遍历字符访问系统敏感文件。 - 分片大小可以根据文件类型和服务器性能调整，视频文件可以适当增大分片大小减少IO次数。
- 如果需要支持断点续传，可以结合请求头中的
Range字段，计算需要读取的文件起始和结束位置，只返回对应片段的内容。 - 响应式流的错误需要做好全局异常处理，避免流异常导致服务不可用。
效果验证
启动Quarkus应用后，访问http://127.0.0.1:8080/preview/test.mp4，浏览器会直接加载视频内容并支持进度条拖动预览，整个过程中服务内存占用保持稳定，不会因为文件体积增大而出现明显波动。同时多个用户同时请求大文件预览时，服务的响应速度也不会出现明显下降。