【Java NIO 简例】非阻塞服务端

原文：《Java NIO: Non-blocking Server》

GitHub 上的实例代码

https://github.com/jjenkov/java-nio-server

 

非阻塞IO管道

非阻塞IO管道是一系列组件的链接。简化结构如下（读写都适用）：

 

Component 利用 Selector 来监测是否有 Channel 就绪可供读取数据；
Component 读取这些输入数据，并基于此生成输出数据；
最后将这些输出数据写到另一个 Channel 中。

 

非阻塞IO管道并不一定同时可读可写。它可以 只读 或 只写。

 

在实际项目中，一个管道可能会有对个对应的 Component 来处理输入的数据；管道的长度也会因业务需要而不同。

 

从 Channel 到 Selector，从 Selector 到 Component 的箭头其实是 Component 在处理 Selector 与 Channel，而不是 Channel 主动将数据推送到 Selector。

 

IO管道：非阻塞 vs 阻塞

非阻塞 与 阻塞IO 的最大区别在于对 channel/stream 的读写方式的不同。

在典型的读取stream数据场景中，可认为有一个 Messager Reader 将stream中的数据分块读出。

 

在阻塞式IO中，可以从一个 InputStream 读取数据；如果 InputStream 中的数据尚未就绪，这个操作将一直阻塞，直到有数据可读。这导致 Messager Reader 是阻塞式的。

阻塞式的 stream 简化了 Message Reader 的实现。因为无需处理stream中无数据就绪，或只有部分数据就绪的情况。

阻塞式写入stream的操作也类似，无需处理只有部分数据写入stream的情况。

 

阻塞式IO管道的缺点

虽然阻塞式 Message Reader 更容易实现，但它需要为每个stream创建一个线程来处理数据。
因为stream在数据就绪前会一直阻塞。这导致单个线程无法在 “尝试读取某个stream，但stream数据未就绪” 的情况下转而去读取另一个stream。因为一个线程一旦尝试去stream读数据，它将被阻塞，直到stream中数据就绪。

 

对于一批来自客户端的并发连接，服务端必须为吗，每个连接创建一个线程进行处理。如果在任一时刻，并发连接数只有几百，也不会有什么问题。但如果是上百万的并发连接，这种阻塞式的设计就无法良好运行。每个线程的线程栈会消耗320KB（32位JVM）或1024KB（64位JVM）的内存。1百万个线程消耗将近1TB的内存！这还没算上服务器用于处理数据的内存。

 

为了降低线程数量，许多服务端都使用了 线程池 的设计。
来自客户端的连接被维护在一个队列中。一旦线程池中有空闲线程，就从队列中取一个连接，由该线程处理。简化结构如下：

 

但是这种设计要求各客户端连接所发送的数据量较为合理，服务端可以较快处理完并返回。否则出现线程池中所有线程长时间被阻塞情况，也就意味着服务端响应慢，或根本就无法响应。

 

在某些服务端设计中，为了减轻服务无响应的状况，会为线程池的线程数量设置一个弹性的值。例如，如果当前线程池中已无空闲线程，线程池可能会动态创建一些新的线程来处理新的客户端连接。这种解决方案意味着在出现服务无响应之前可以扛住更多耗时的客户端连接。但是，请记住，可以同时拥有线程数的上限依然存在。即，它无法处理一百万个耗时的并发客户端连接。

 

基本的非阻塞IO管道设计

一个非阻塞IO管道可以用单个线程读取多个stream的数据。这要求这些stream可以非阻塞模式工作。在非阻塞模式下，当尝试读取stream中的数据时，可能得到数据，也可能没有数据。读到数据说明在读取前stream中有数据就绪了；没读到数据说明stream中尚未有就绪的数据。

 

为避免（业务的代码主动）检查stream是否有数据就绪可读，我们可以使用 Selector。可以将一个或多个 SelectableChannel 实例注册到一个 Selector 上（监听Read事件）。当调用 Selector 的 select() 或 selectNow() 方法时，Selector 只会返回有数据就绪可读的 channel。简化结构如下：

 

 

读取部分Message

当我们从一个 SelectableChannel 读到一个数据块时，我们并不知道这个数据块所包含的数据量是否刚好是业务意义上的一条Message，亦或是少于或多于。

 

处理部分（不完整）Message有两个挑战：

• 检测数据块是否包含了一个完整的Message；
• 在一条Message的后部分数据到达前，如何处理已到达的这部分不完整的Message。

 

检测完整的Message 要求 MessageReader 查看数据块中的数据内容，以确定是否含有完整的Message。如果数据块中存在一条或多条Message，则这些具有相对独立业务意义的Message可以被传递到管道的下一步处理。因为检测完整Message的操作会被重复许多次，所以这个检测操作必须尽可能得快。

 

任何时候，如果数据块中出现了部分（非完整）的Message，需要先将这部分数据缓存下来，等其它部分数据到达后再处理。

检测完整Message 和 缓存非完整Message 都是 MessageReader 的职责。为了避免混淆不同 channel 的数据，可以给每个 channel 配一个 MessageReader。简化结构如下：

 

一旦从 Selector 获得一个有数据就绪可读的 channel，Message Reader 将尝试从 channel 读取数据并分割成业务意义上的Message

 

Message Reader 肯定是与特定的（通信）协议相关的（甚至包括业务逻辑）。Message Reader 必须知道它所读消息的格式。如果想要让服务端的实现能跨协议复用，它必须有能力使得 Messager Reader 是以类似插件的形式存在的。比如，有一个 Message Reader 工厂根据协议配置参数创建 Message Reader 实例。

 

缓存不完整的Message

我们已经确定“在收到Message完整数据前，缓存不完整的那部分数据”是 Message Reader 的职责。现在需要明确如何实现不完整Message的缓存。

在设计上我们需要考虑两点：

• 复制的数据越少越好。拷贝得越多，性能越差。
• 完整Message应存储在连续的字节序列中，以方便解析。

 

每个 Message Reader 对应一个缓冲区

显然，不完整的Message需要被暂存于某种缓冲区中。最直接的实现是，在每个 Message Reader 中添加一个 buffer。但是，这个 buffer 该多大呢？它需要大到足以放下最大允许的Message。所以，如果最大允许的Message是1MB，那么每个 Message Reader 中的buffer至少是1MB。

当连接数达到上百万时，每个连接配1MB buffer 的方案就不管用了。一百万个连接，要耗费将近1TB内存！如果Message最大是16MB呢？128MB呢？

 

可调整大小的Buffer

另一种选择是在每个 Messsage Reader 中实现一个可调整大小的buffer。该buffer开始时比较小，如果后续message数据量大于buffer容量，buffer将被扩展。这样就不必为每个连接分配1MB的buffer。每个连接都只将耗费各自接收下一条Message的内存。

有多种方法来实现一个可调整大小的buffer。它们都有各自的优缺点，我将在后续章节讨论。

 

以复制数据的方式调整大小

先创建一个较小的buffer，如 4KB。如果buffer放不下某条Message，再分配一个更大的buffer，如 8KB，并将原buffer中的数据拷贝到新buffer中。

优点：一条Message的所有数据都被存放在单个连续的字节数组中。这可以使Message解析工作更简单。

缺点：对于数据量较大的Message，需要拷贝大量数据。

 

为了减少数据拷贝的开销，你可以研究你的业务系统，找到可以减少数据拷贝量的buffer大小。如，你可能会发现绝大多数message都小于4KB，因为它们只包含数据量很少的请求与响应。这意味着，buffer的初始大小应是4KB。

然后你可能会发现超过4KB的message通常都包含了一个文件。然后你可能会注意到该业务系统中绝大多数文件都小于128KB。所以第二个buffer大小应该是12bKB。

最终，你可能发现一旦一条Message超过128KB，其大小就没有什么规律了。所以最后一个buffer的应该就是最大Message的大小。

 

有了以上三个基于业务系统Message大小的buffer容量，你可以减少一些数据拷贝的开销。4KB以下的Message永远都不会被拷贝。对于一百万并发连接来说，这将消耗近4GB。这对于现在（2015年）的绝大多数服务器来说是可行的。大小介于4KB和128KB的Message只会被拷贝一次，而且只有4KB那部分数据需要被拷贝（从4KB的buffer拷贝到128KB的buffer）。大小介于128KB和Message数据量最大值之间的Message会被拷贝两次。第一次是拷贝4KB，第二次是拷贝128KB。所以对于这些大数据量的Message来说，每条总共需拷贝132KB的数据。假设超过128KB的Message数据量不多，那么该方案也是可接受的。

 

一旦一条Message被处理完毕，需要释放为期分配的相关内存。这样，来自同一个连接的下一条Message将又从最小的buffer开始。因为有必要确保连接之间可以更有效地共享内存。极少会发生所有连接在同一时刻都需要一个大buffer。

 

我已完成一个关于支持数组大小可调整的内存buffer如何实现的指南：Resizable Arrays。其中包括一个展示代码实现的GitHub仓库链接。

 

以追加数据的方式调整大小

让buffer包含多个array。当需要增加buffer的容量时，只需再分配一个字节数组，并将新的数据填写到该数组中。

有两种方法来实现这样的buffer。一种方法是分配一个额外的数组来维护这些存放Message数据的数组。另一种方法是从一个共享的大字节数组中分配出多个（内存）分片，并维护一个分配给该buffer的分片的列表。我个人稍倾向于分片的方案。但两种方案的差别很小。

优点：无需额外拷贝数据。所有 socket（channel）的数据可以直接被拷贝到某个数组或分片。

缺点：Message解析更困难。因为数据不是存放在单个连续的字节数组中，Message解析器需要检查数据数组的末尾数据，还需检查维护这些数据数组的列表末尾数据。既然需要检查Message的末尾数据，这种模式就不容易实现了。

 

TLV 编码的Message（Type、Length、Value）

某些协议的Message是以TLV（Type、Length、Value）的格式编码的。这意味着，当收到一条Message时，该Message的数据长度在其最前面。这样你就能立即知道该为这一整条Message分配多少内存。

TLV编码使得内存管理更简单。你可以立即知道需为Message分配多少内存。buffer中末尾的内存不会因为只存了部分Message数据而被浪费。

缺点：在一条Message的所有数据到达前，就为其分配所需的所有内存。少数包含大Message的慢连接可能会占用所有可用内存，导致服务无响应。

 

针对上述问题的一个解决方法是采用包含多个TLV字段的Message格式。这样，内存的分配就是针对每个单独的字段，而非整个Message；只有当收到某个字段的数据时，才为该字段分配内存。但是一个大数据量的字段仍可能像大数据量的Message那样对系统产生上述负面效应。

另一种解决方法是将那些超过规定时限（如，10~15秒）仍未完全到达的Message置为“超时”。这样可以让系统在同时收到大量大Message的极端情况下恢复过来。但这还是会导致服务有一段时间无响应。而且故意的DoS（Denial of Service）攻击仍会耗尽你的服务器内存。

 

TLV编码有多种变异。一个Message字段具体用了多少个字节来表示其数据类型和长度，在不同TLV编码中都可能不同。某些TLV编码将字段的长度放在最前，类型其次，最后是数据值。也有的TLV编码变种以不同的顺序组织这些信息。

TLV编码可以使内存管理更简单也是 HTTP 1.1 协议糟糕的原因之一。这也是 HTTP 2.0 尝试修复的问题之一。这也是为什么我们自行设计的网络协议 VStack.co Project 采用了TLV编码。

 

写入部分（非完整的）Message

往非阻塞式IO管道写入数据也是一项挑战。当你调用 非阻塞模式Channel 的 write(ByteBuffer) 方法时，无法保证ByteBuffer中有多少字节正被写入。write(ByteBuffer)方法会返回被写入字节的数量，所以可以追踪已完成写入的字节数。这也是挑战所在：追踪已被写入的部分Message，并确保Message的所有字节都最终发送出去。

为了管理向 Channel 写入部分 Message，我们可以创建一个 Message Writer。如同Message Reader，每个被将写入数据的Channel都需要有一个 Message Writer。在每个Message Writer中，我们持续追踪当前正在被写入的Message已经完成写入的具体字节数。

为防止更多Message到达Message Writer，超过其可以直接写到 Channel 的数量，这些Message需要在Message Writer 内部进行排队。这样Message Writer就能以它最快的速度将Message写入Channel。

简化结构如下：

 

为了使Message Writer能将那些已发送了部分数据的Message全部数据都发出去，Message Writer 需要时不时地被调用，以发送更多数据。

如果你有许多连接，将需要许多Message Writer实例。检查一百万个Message Writer实例是否能写数据是很慢的。首先，可能有许多Message Writer实例没有任何Message可供发送。我们也不想检查这些实例。其次，并不是所有的Channel实例已就绪可被写入数据。我们也不想在尝试往这些Channel写数据上浪费时间。

为检查一个Channel是否已就绪可写，你可以将其注册到一个 Selector 上。但是我们不想把所有 Channel 实例都注册到一个 Selector 上。想象一下，有一百万个连接，其中绝大多数是空闲的，且所有这些连接都被注册到了一个 Selector 上。那么当你调用 Selector.select() 方法时，绝大多数Channel实例都是就绪可写的（记住，它们绝大多数是空闲的）。这样你不得不在Message Writer中检查这些连接是否有数据要写。

为了避免检查所有Message Writer实例是否有Message要发送，我们采用了分两步的方法：

1. 当一个Message被送入Message Writer，该Message Writer将其相关的Channel注册到Selector中（如果还未注册）。
2. 当你的服务器有空闲时，检查Selector中哪些Channel已就绪可写。对每个就绪可写的Channel，触发它相关的Message Writer向Channel中写数据。如果Message Writer已将所有相关Message写入其Channel，该Channel将被从Selector中注销。

此两步方法确保了只有那些需要被写入数据的Channel才会被真正地注册到Selector上。

 

合在一起

如你所见，非阻塞式服务端需要时不时地检查接收到的数据，以发现是否已收到新的完整Message。服务端可能需要检查多次直到收到一条（或多条）完整的Message。只检查一次是不够的。

同样的，非阻塞式服务端还需要时不时地检查是否有数据需要发送。如果有，服务端需要检查是否有相关连接已就绪可写。只在Message刚进入队列时检查是不够的，因为Message可能只有部分数据被发出。

总而言之，非阻塞式服务端需要有规律地执行三个“通道”：

• 读通道：检查是否有新收到的数据。
• 处理通道：处理已收到的完整Message。
• 写通道：是否有Message需要发送且可以发送到相关连接。

这三个“管道”在循环中被重复执行。你也许能优化这个执行。例，如果没有Message在写队列中，你可以跳过“写通道”。或者，如果没有收到新的完整Message，你可以跳过“处理通道”。

简化结构如下：

 

如果你仍觉得这优点复杂，记得去查看GitHub仓库：
https://github.com/jjenkov/java-nio-server

也许看到实际代码可以帮你理解如何实现它。

 

服务端线程模型

GitHub仓库中的非阻塞服务端实现使用了一个包含2个线程的线程模型。第一个线程从ServerSocketChannel接受来自客户端的连接。第二个线程处理已接受的连接，也就是读取Message、处理Message、向连接回写响应。简化结构如下：