3

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

问题：如何使用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

4

4 回答 4

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;
}
}

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

于 2012-11-14T20:13:58.433 回答
0

在我看来，最简单的方法是将此代码拆分为自己的方法，然后让 CompletionHandler 在bytesRead != -1. 这样，您可以分离出代码的职责，并避免在异步读取运行时“忙等待”或使用 Thread.sleep() 的必要性。当然，您也可以添加一个案例，何时bytesRead == -1对已读入的数据执行某些操作。

于 2012-11-10T22:35:34.923 回答
0

我不会做任何延长回调方法的事情failed，completed因为它们在不受您控制的线程上运行。

我了解您想继续侦听套接字中的新字节，即使流已到达其末尾（bytesRead == -1）。将read方法放在一个while(true)循环中。在其中，监听由and方法synchronized设置的字段。让我们称之为。failedcompletedmyBytesRead

为了能够停止无限读取，请将 替换为while(true)其他synchronized条件。

private static final BYTES_READ_INIT_VALUE = Integer.MIN_VALUE;
private static final BYTES_READ_COMPLETED_VALUE = -1;
private static final BYTES_READ_FAILED_VALUE = -2;
private Integer myBytesRead = BYTES_READ_INIT_VALUE;

private void setMyBytesRead(final Integer bytesRead) {
    synchronized(myBytesRead) {
        this.myBytesRead = bytesRead;
    }
}

private Integer getMyBytesRead() {
    synchronized(myBytesRead) {
        return myBytesRead;
    }
}

...

// in your method
while (true) {
    final int lastBytesRead = getMyBytesRead();
    if (lastBytesRead == BYTES_READ_FAILED_VALUE) {
        // log failure and retry?
    } else if (lastBytesRead != BYTES_READ_COMPLETED_VALUE) {
        // Thread.sleep( a while ); to avoid exhausting CPU
        continue;
    }

    // else lastBytesRead == BYTES_READ_COMPLETED_VALUE and you can start a new read operation
    client.read(readTo, readTo, new CompletionHandler<Integer, ByteBuffer>() {
            public void completed(Integer bytesRead, ByteBuffer buffer) {
                setMyBytesRead(bytesRead);
                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 {
                    setMyBytesRead(BYTES_READ_FAILED_VALUE);
                    client.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    );
}
于 2012-11-10T23:12:37.063 回答
0

我最初的尝试：

package com.example;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;

public class LoopingReader implements Callable<String> {
    final AsynchronousSocketChannel client;
    final String responseTerminator;
    final StringBuilder responseBuilder;

    LoopingReader(
        AsynchronousSocketChannel client,
        String responseTerminator
    ) {
        this.client = client;
        this.responseTerminator = responseTerminator;

        responseBuilder = new StringBuilder();
    }

    public String call() {
        boolean doLoop;
        do {
            int bytesRead = executeIteration();//blocking
            boolean didReachEndOfStream = bytesRead == -1;
            boolean didEncounterResponseTerminator = responseBuilder.indexOf(responseTerminator) != -1;

            doLoop =  !didReachEndOfStream && !didEncounterResponseTerminator;
        } while(doLoop);
        return responseBuilder.toString();
    }

    int executeIteration() {
        final ByteBuffer buffer = ByteBuffer.allocate(256);//use pool here
        final int bytesRead;
        try {
            bytesRead = client.read(buffer).get();
        } catch (InterruptedException | ExecutionException e) {
            throw new IllegalStateException("Failed to read", e);
        }
        decodeAndAppend(buffer);
        return bytesRead;
    }

    void decodeAndAppend(ByteBuffer buffer) {
        buffer.flip();
        responseBuilder.append(StandardCharsets.US_ASCII.decode(buffer));
    }
}
于 2012-11-14T00:53:04.473 回答