java - 如何使用 Java NIO 限制从 SocketChannel InputStream 一次读取一行

Question

我正在尝试编写一个 Websockets 客户端和服务器。最初连接是 HTTP，Websockets 握手使用 HTTP header 来指示连接需要升级到新协议(protocol)。

我想从 SocketChannel 读取 HTTP header 集，如果指示升级，则切换到不同的库来处理 Websocket，从那时起，以完全不同的方式处理 SocketChannel 流，将其作为一组帧而不是用\r\n 分隔的行。

我知道我可以将任意数量的字节读取到 ByteBuffer 中，但是 Websockets 帧可能已通过握手发送，并且我不想在这些代码段之间传递半消耗的缓冲区。我想要的是从套接字仅读取直到并包括序列“\r\n\r\n”的数据。除此之外我想保留在 SocketChannel 对象输入流中的任何数据。

推荐的方法是什么？从 SocketChannel 获取输入流并将其包装在缓冲读取器中？这是否可以与 NIO 正确交互，特别是非阻塞使用？一旦检测到空行，我是否可以从输入流中删除缓冲读取器，并且当 channel 传递到 Websockets 代码时仍然拥有所有可用的帧数据？

或者也许我需要逐字节读取(或者如果某些目标“\r\n\r\n”字符出现在 block 的末尾，则读取具有较小缓冲区的 4 字节 block )并构建我的 header 字符串就是这样。

或者，如果直接分配缓冲区，则操作标记、限制和位置的某种组合可能允许输入流取回之前读入 ByteBuffer 的数据。

如有任何建议，我们将不胜感激。

Answer 1

我建议使用 Apache Mina 或 Grizzly 之类的东西。两者都允许您封装问题的协议(protocol)方面，因此您只需处理可使用的数据。

但是，如果您想要一种快速而肮脏的方式:但基本思想是，您需要在数据传入时读取数据。如果它不容易使用，我通常会为选择器中的 SelectionKey 创建一些可附加结构(对于简单的 StringBuilder )。每次读取后，我会将数据附加到构建器，如果检测到可用的 header ，则将其从缓冲区中切片并将其向上传递(最好在工作线程上)。继续这样做，上游的任何东西都应该能够做出相应的 react 。希望有帮助。

所以通常你有这样的结构:

ByteBuffer reUsableBuffer = ByteBuffer.allocateDirect(5120);
Selector selector = Selector.open();
ServerSocketChannel channel = .. // wherever you get it from 
channel.register(selector, SelectionKey.OP_ACCEPT);
Executor executor = Executors.newThreadPoolExecutor();
while(selector.isOpen()) { 
 int numKey = selector.select();
 for (SelectionKey key: selector.selectedKeys()) {
    if (key.isAcceptable()) {
             /// Sort of included for completeness but you get the idea
           ServerSocketChannel server = (ServerSocketChannel)key.channel();
           SocketChannel channel = server.accept();
           channel.register(selector, SelectionKey.OP_READ | Selection.OP_WRITE, new StringBuilder());
    }    if (key.isReadable()) {
          // READ the data
          reUsableBuffer.clear();
          // You have to keep track of previous state.
          // NIO makes no guarantees of anything
          StringBuilder builder = key.attachment();
          SocketChannel socketChannel = (SocketChannel)key.channel();
          int readCount = socketChannel.read(reUsableBuffer);
          if (readCount > 0) {
             reUsableBuffer.flip();
             byte[] subStringBytes = new byte[readCount];
             reUsableBuffer.read(subStringBytes);
             // Assuming ASCII (bad assumption but simplifies the example)
             builder.append(new String(substringBytes));

             Command[] commands = removeCommands(builder);
             // Deal with your commands in some async manor defined by you
             executor.execute(new Task(commands));
          }
        }
        selector.selectedKeys().clear(); } ....

    }   

//
// Parse out the commands and return them, also remove traces of them in the
// the builder, such that for a string, "COMMAND, COMMAND, COM"
// an array of 2 should be returned with a left over buffer of "COM"
public Command[] parseCommands(StringBuilder s) { ... }