java - Java AsyncHttpClient:从 LazyResponseBodyPart 写入 AsynchronousFileChannel 时文件损坏
问题描述
我将AsyncHttpClient 库用于异步非阻塞请求。我的情况:将数据写入文件,因为它是通过网络接收的。
对于从远程主机下载文件并保存到文件,我使用默认值ResponseBodyPartFactory.EAGER
,AsynchronousFileChannel
以免在数据到达时阻塞 netty 线程。但正如我的测量结果显示的那样,与LAZY
Java 堆中的内存消耗相比,它增加了很多倍。
所以我决定直接去LAZY
,但没有考虑文件的后果。
此代码将有助于重现问题。:
public static class AsyncChannelWriter {
private final CompletableFuture<Integer> startPosition;
private final AsynchronousFileChannel channel;
public AsyncChannelWriter(AsynchronousFileChannel channel) throws IOException {
this.channel = channel;
this.startPosition = CompletableFuture.completedFuture((int) channel.size());
}
public CompletableFuture<Integer> getStartPosition() {
return startPosition;
}
public CompletableFuture<Integer> write(ByteBuffer byteBuffer, CompletableFuture<Integer> currentPosition) {
return currentPosition.thenCompose(position -> {
CompletableFuture<Integer> writenBytes = new CompletableFuture<>();
channel.write(byteBuffer, position, null, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
writenBytes.complete(result);
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
writenBytes.completeExceptionally(exc);
}
});
return writenBytes.thenApply(writenBytesLength -> writenBytesLength + position);
});
}
public void close(CompletableFuture<Integer> currentPosition) {
currentPosition.whenComplete((position, throwable) -> IOUtils.closeQuietly(channel));
}
}
public static void main(String[] args) throws IOException {
final String filepath = "/media/veracrypt4/files/1.jpg";
final String downloadUrl = "https://m0.cl/t/butterfly-3000.jpg";
final AsyncHttpClient client = Dsl.asyncHttpClient(Dsl.config().setFollowRedirect(true)
.setResponseBodyPartFactory(AsyncHttpClientConfig.ResponseBodyPartFactory.LAZY));
final AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get(filepath), StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.CREATE);
final AsyncChannelWriter asyncChannelWriter = new AsyncChannelWriter(channel);
final AtomicReference<CompletableFuture<Integer>> atomicReferencePosition = new AtomicReference<>(asyncChannelWriter.getStartPosition());
client.prepareGet(downloadUrl)
.execute(new AsyncCompletionHandler<Response>() {
@Override
public State onBodyPartReceived(HttpResponseBodyPart content) throws Exception {
//if EAGER, content.getBodyByteBuffer() return HeapByteBuffer, if LAZY, return DirectByteBuffer
final ByteBuffer bodyByteBuffer = content.getBodyByteBuffer();
final CompletableFuture<Integer> currentPosition = atomicReferencePosition.get();
final CompletableFuture<Integer> newPosition = asyncChannelWriter.write(bodyByteBuffer, currentPosition);
atomicReferencePosition.set(newPosition);
return State.CONTINUE;
}
@Override
public Response onCompleted(Response response) {
asyncChannelWriter.close(atomicReferencePosition.get());
return response;
}
});
}
在这种情况下,图片被破坏了。但是,如果我使用FileChannel
而不是AsynchronousFileChannel
,在这两种情况下,文件都会正常显示。DirectByteBuffer
使用(以防万一LazyResponseBodyPart.getBodyByteBuffer()
)和工作时会有任何细微差别AsynchronousFileChannel
吗?
如果一切正常,我的代码可能有什么问题EAGER
?
更新
正如我所注意到的,如果我使用LAZY
,例如,我
Thread.sleep (10)
在方法中添加该行onBodyPartReceived
,如下所示:
@Override
public State onBodyPartReceived(HttpResponseBodyPart content) throws Exception {
final ByteBuffer bodyByteBuffer = content.getBodyByteBuffer();
final CompletableFuture<Integer> currentPosition = atomicReferencePosition.get();
final CompletableFuture<Integer> newPosition = finalAsyncChannelWriter.write(bodyByteBuffer, currentPosition);
atomicReferencePosition.set(newPosition);
Thread.sleep(10);
return State.CONTINUE;
}
该文件以非损坏状态保存到磁盘。
据我了解,原因是在这 10 毫秒内,异步线程AsynchronousFileChannel
设法从 this 将数据写入磁盘DirectByteBuffer
。
事实证明,由于这个异步线程使用这个缓冲区与 netty 线程一起写入,所以文件被破坏了。
如果我们用 看一下源代码EagerResponseBodyPart
,那么我们将看到以下内容
private final byte[] bytes;
public EagerResponseBodyPart(ByteBuf buf, boolean last) {
super(last);
bytes = byteBuf2Bytes(buf);
}
@Override
public ByteBuffer getBodyByteBuffer() {
return ByteBuffer.wrap(bytes);
}
因此,当一条数据到达时,它会立即存储在字节数组中。然后我们可以安全地将它们包装在 HeapByteBuffer 中并传输到文件通道中的异步线程。
但是如果你看代码LazyResponseBodyPart
private final ByteBuf buf;
public LazyResponseBodyPart(ByteBuf buf, boolean last) {
super(last);
this.buf = buf;
}
@Override
public ByteBuffer getBodyByteBuffer() {
return buf.nioBuffer();
}
如您所见,我们实际上通过方法调用在异步文件通道线程中使用netty ByteBuff
(在这种情况下总是)PooledSlicedByteBuf
nioBuffer
在这种情况下我该怎么办,如何DirectByteBuffer
在不将缓冲区复制到 Java 堆的情况下安全地传入异步线程?
解决方案
我与AsyncHttpClient
.
可以看这里
主要问题是我不使用 netty ByteBuf 方法retain
和release
. 最后,我想到了两个解决方案。
第一:按顺序将字节写入文件,跟踪位置为CompletableFuture
.
定义包装类AsynchronousFileChannel
@Log4j2
public class AsyncChannelNettyByteBufWriter implements Closeable {
private final AtomicReference<CompletableFuture<Long>> positionReference;
private final AsynchronousFileChannel channel;
public AsyncChannelNettyByteBufWriter(AsynchronousFileChannel channel) {
this.channel = channel;
try {
this.positionReference = new AtomicReference<>(CompletableFuture.completedFuture(channel.size()));
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}
public CompletableFuture<Long> write(ByteBuf byteBuffer) {
final ByteBuf byteBuf = byteBuffer.retain();
return positionReference.updateAndGet(x -> x.thenCompose(position -> {
final CompletableFuture<Integer> writenBytes = new CompletableFuture<>();
channel.write(byteBuf.nioBuffer(), position, byteBuf, new CompletionHandler<Integer, ByteBuf>() {
@Override
public void completed(Integer result, ByteBuf attachment) {
attachment.release();
writenBytes.complete(result);
}
@Override
public void failed(Throwable exc, ByteBuf attachment) {
attachment.release();
log.error(exc);
writenBytes.completeExceptionally(exc);
}
});
return writenBytes.thenApply(writenBytesLength -> writenBytesLength + position);
}));
}
public void close() {
positionReference.updateAndGet(x -> x.whenComplete((position, throwable) -> {
try {
channel.close();
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}));
}
}
事实上,这里可能不会有一个AtomicReference
,如果录制发生在一个线程中,如果来自多个线程,那么我们需要认真对待同步。
和主要用途。
public static void main(String[] args) throws IOException {
final String filepath = "1.jpg";
final String downloadUrl = "https://m0.cl/t/butterfly-3000.jpg";
final AsyncHttpClient client = Dsl.asyncHttpClient(Dsl.config().setFollowRedirect(true)
.setResponseBodyPartFactory(AsyncHttpClientConfig.ResponseBodyPartFactory.LAZY));
final AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get(filepath), StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.CREATE);
final AsyncChannelNettyByteBufWriter asyncChannelNettyByteBufWriter = new AsyncChannelNettyByteBufWriter(channel);
client.prepareGet(downloadUrl)
.execute(new AsyncCompletionHandler<Response>() {
@Override
public State onBodyPartReceived(HttpResponseBodyPart content) {
final ByteBuf byteBuf = ((LazyResponseBodyPart) content).getBuf();
asyncChannelNettyByteBufWriter.write(byteBuf);
return State.CONTINUE;
}
@Override
public Response onCompleted(Response response) {
asyncChannelNettyByteBufWriter.close();
return response;
}
});
}
第二种解决方案:根据接收到的字节大小跟踪位置。
public static void main(String[] args) throws IOException {
final String filepath = "1.jpg";
final String downloadUrl = "https://m0.cl/t/butterfly-3000.jpg";
final AsyncHttpClient client = Dsl.asyncHttpClient(Dsl.config().setFollowRedirect(true)
.setResponseBodyPartFactory(AsyncHttpClientConfig.ResponseBodyPartFactory.LAZY));
final ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
final AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get(filepath), new HashSet<>(Arrays.asList(StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.CREATE)), executorService);
client.prepareGet(downloadUrl)
.execute(new AsyncCompletionHandler<Response>() {
private long position = 0;
@Override
public State onBodyPartReceived(HttpResponseBodyPart content) {
final ByteBuf byteBuf = ((LazyResponseBodyPart) content).getBuf().retain();
long currentPosition = position;
position+=byteBuf.readableBytes();
channel.write(byteBuf.nioBuffer(), currentPosition, byteBuf, new CompletionHandler<Integer, ByteBuf>() {
@Override
public void completed(Integer result, ByteBuf attachment) {
attachment.release();
if(content.isLast()){
try {
channel.close();
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}
}
@Override
public void failed(Throwable exc, ByteBuf attachment) {
attachment.release();
try {
channel.close();
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}
});
return State.CONTINUE;
}
@Override
public Response onCompleted(Response response) {
return response;
}
});
}
在第二种方案中,因为我们不等到一些字节被写入文件,AsynchronousFileChannel
所以可以创建很多线程(如果你使用Linux,因为Linux没有实现非阻塞异步文件IO。在Windows中,情况是好多了)。
正如我的测量结果所示,在写入慢速 USB 闪存的情况下,线程数可能达到数万,因此您需要通过创建ExecutorService
并传输到AsynchronousFileChannel
.
第一种和第二种解决方案有明显的优缺点吗?我很难说。也许有人可以告诉什么更有效。
推荐阅读
- kubernetes - Kubernetes:你如何列出在 master 上运行的组件?
- python - Python:所有相同的值用“for~in”进入数组
- javascript - 无法通过 nodejs express 从 HTML/JS 客户端获取 ASP.NET Core 3 中的二进制数据
- c# - 检查是否调用了 WCF 服务
- mysql - sql如何选择具有两行或多行的组的倒数第二行
- networking - Taskmanager 无法连接到 Jobmanager [找不到任何不是环回或链路本地的 IPv4 地址。使用本地主机地址。]
- python - 如何使用 sklearn.decomposition FactorAnalysis 在 python 中获取因子载荷
- javascript - 打字稿:如何在构造函数之外初始化类属性
- ios - 使用 Swift iOS 在 SFSafariViewController 中配置浏览器缓存
- python - 在元素计数不同于 1 的组上过滤 DataFrame