java - 如何在循环或递归中使用 AsynchronousSocketChannel#read？

Question

我找到了一个 related question ，但它并不是特别有用，因为它没有提供完整的示例。

问题:如何使用AsynchronousSocketChannel用于使用固定大小的缓冲区读取未知长度的数据

第一次尝试(读取一次):

final int bufferSize = 1024;
final SocketAddress address = /*ip:port*/;
final ThreadFactory threadFactory = Executors.defaultThreadFactory();
final ExecutorService executor = Executors.newCachedThreadPool(threadFactory);
final AsynchronousChannelGroup asyncChannelGroup = AsynchronousChannelGroup.withCachedThreadPool(executor, 5);
final AsynchronousSocketChannel client = AsynchronousSocketChannel.open(asyncChannelGroup);
client.connect(address).get(5, TimeUnit.SECONDS);//block until the connection is established

//write the request
Integer bytesWritten = client.write(StandardCharsets.US_ASCII.encode("a custom request in a binary format")).get();

//read the response
final ByteBuffer readTo = ByteBuffer.allocate(bufferSize);
final StringBuilder responseBuilder = new StringBuilder();
client.read(readTo, readTo, new CompletionHandler<Integer, ByteBuffer>() {
        public void completed(Integer bytesRead, ByteBuffer buffer) {
            buffer.flip();
            responseBuilder.append(StandardCharsets.US_ASCII.decode(buffer).toString());
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        
        public void failed(Throwable exc, ByteBuffer attachment) {
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
);
asyncChannelGroup.awaitTermination(5, TimeUnit.SECONDS);
asyncChannelGroup.shutdown();
System.out.println(responseBuilder.toString());

我需要做哪些更改才能干净地实现在bytesRead != -1(即到达流末尾)时连续读入缓冲区？

Answer 1

这是我最终所做的简化版本(使用 Guava 的 ListenableFuture ):

class SomeUtilClass {
public interface Processor<T> {
    boolean process(Integer byteCount, ByteBuffer buffer);
    T result();
}
public static <T> ListenableFuture<T> read(
    final AsynchronousSocketChannel delegate,
    final Processor<T> processor,
    ByteBuffer buffer
) {
    final SettableFuture<T> resultFuture = SettableFuture.create();
    delegate.read(buffer, buffer, new Handler<T, Integer, ByteBuffer>(resultFuture) {
        public void completed(Integer bytesRead, ByteBuffer buffer) {
            buffer.flip();
            if(processor.process(bytesRead, buffer)) {
                buffer.clear();
                delegate.read(buffer, buffer, this);
            } else {
                resultFuture.set(processor.result());
            }
        }
    });
    return resultFuture;
}
}

进一步改进包括使用 Commons Pool的 ByteBuffer