我使用 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
时,是否存在细微差别?如果一切正常,为什么我的代码会有问题,如果我使用?
更新
我注意到,如果我使用
LAZY
,并在方法onBodyPartReceived
中添加行Thread.sleep(10)
,就像这样: @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
中的异步线程成功地从 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堆中?
BodyDeferringAsyncHandler
来简化生活呢? - Mạnh Quyết NguyễnBodyDeferringAsyncHandler
在内存消耗方面与我使用EAGER
的示例没有区别,因为BodyDeferringAsyncHandler
使用getBodyPartBytes
方法。我不确定,但可能在使用BodyDeferringAsyncHandler
时,线程会在写入OutputStream
时阻塞。 - Peter KozlovskyAsynchronousFileChannel
时,这是一个阻塞操作,因为Java在Linux中没有实现真正的文件AIO。当数据通过网络到达时,AsyncHttpClient
线程不会被阻塞,当数据到达onBodyPartRecieved
方法时,我们不应该阻塞netty线程。 然而,我们已经偏离了最初的问题。 - Peter Kozlovsky