从经典多线程到 java.nio 异步/非阻塞服务器

Posted 2023-03-06

【中文标题】从经典多线程到 java.nio 异步/非阻塞服务器

【英文标题】：From classic multithreaded to java.nio asynchronous/non-blocking server

【发布时间】：2013-03-11 20:31:47

【问题描述】：

我是一款在线游戏的主要开发者。 玩家使用特定的客户端软件通过 TCP/IP（TCP，而不是 UDP）连接到游戏服务器

目前，服务器的架构是经典的多线程服务器，每个连接一个线程。 但是在高峰时段，当经常有 300 或 400 个连接的人时，服务器会变得越来越慢。

我想知道，如果切换到 java.nio.* 异步 I/O 模型，用很少的线程管理很多连接，性能是否会更好。 在 Web 上查找涵盖此类服务器架构基础知识的示例代码非常容易。然而，经过几个小时的谷歌搜索，我没有找到一些更高级问题的答案：

1 - 该协议是基于文本的，而不是基于二进制的。客户端和服务器交换以 UTF-8 编码的文本行。单行文本代表一个命令，每一行都以 \n 或 \r\n 正确终止。 对于经典的多线程服务器，我有这样的代码：

public Connection (Socket sock) 
this.in = new BufferedReader( new InputStreamReader( sock.getInputStream(), "UTF-8" ));
this.out = new BufferedWriter( new OutputStreamWriter(sock.getOutputStream(), "UTF-8"));
new Thread(this) .start();

然后在运行中，用readLine逐行读取数据。

在文档中，我找到了一个实用类 Channels，它可以从 SocketChannel 创建一个 Reader。但是据说如果 Channel 处于非阻塞模式，则生成的 Reader 将无法工作，这与非阻塞模式必须使用我愿意使用的高性能通道选择 API 的事实相矛盾。所以，我怀疑这不是我想做的正确解决方案。 因此，第一个问题如下：如果我不能使用它，如何有效和正确地处理断行并将本机 java 字符串从/转换为 nio API 中的 UTF-8 编码数据，带有缓冲区和通道？ 我是否必须手动使用 get/put 或在包装的字节数组中？如何从 ByteBuffer 转到以 UTF-8 编码的字符串？我承认不太了解如何使用 charset 包中的类以及它是如何做到这一点的。

2 - 在异步/非阻塞 I/O 世界中，对于本质上一个接一个地依次执行的连续读/写的处理呢？ 例如，典型的基于挑战-响应的登录过程：服务器发送一个问题（特定计算），客户端发送响应，然后服务器检查客户端给出的响应。 答案是，我认为，当然不要为整个登录过程创建一个任务发送到工作线程，因为它很长，有冻结工作线程太长时间的风险（想象一下这种情况：10 个池线程, 10 个玩家同时尝试连接；与已在线玩家相关的任务会延迟到一个线程再次准备好）。

3 - 如果两个不同的线程同时在同一个 Channel 上调用 Channel.write(ByteBuffer) 会发生什么？ 客户可能会收到混淆的线路吗？例如，如果一个线程发送“aaaaa”而另一个线程发送“bbbbb”，客户端是否可以接收“aaabbbbbaa”，或者我是否确保每个线程都以一致的顺序发送？我可以在调用返回后立即修改使用的缓冲区吗？ 或者换一种方式问，我是否需要额外的同步来避免这种情况？ 如果我需要额外的同步，如何知道何时释放锁等，写完成后？ 恐怕答案并不像在选择器中注册 OP_WRITE 那样简单。通过尝试，我注意到我一直都在为所有客户端获得 write-ready 事件，大部分都是提前退出 Selector.select，因为每个客户端每秒只有 3 或 4 条消息要发送，而选择循环每秒执行数百次。因此，从潜在的角度来看，积极等待是非常糟糕的。

4 - 多个线程能否同时在同一个选择器上调用 Selector.select，而不会出现任何并发问题，例如丢失事件、调度两次等？

5 - 事实上，nio 真的像传说中的那么好吗？保留经典的多线程模型是否会很有趣，但不是为每个连接创建一个线程，而是使用更少的线程并循环连接以使用 InputStream.isAvailable 查找数据可用性？这个想法是愚蠢和/或低效的吗？

【问题讨论】：

对于一些示例代码，请查看 Netty 的源代码：github.com/netty/netty 这是一个非常好的库。

【参考方案1】：

1) 是的。我认为您需要编写自己的非阻塞 readLine 方法。另请注意，当缓冲区中有几行或有不完整的行时，可能会发出非阻塞读取信号：

例子：（初读）

 USER foo
 PASS

（二读）

 bar

您需要存储（参见 2）未使用的数据，直到有足够的信息准备好处理它。

 //channel was select for OP_READ
 read data from channel 
 prepend data from previous read
 split complete lines
 save incomplete line
 execute commands

2) 您需要保留每个客户端的状态。

    Map<SocketChannel,State> clients = new HashMap<SocketChannel,State>();

当一个通道被连接时，put 一个新的状态进入地图

    clients.put(channel,new State());

或将当前状态存储为SelectionKey 的the attached object。

然后，在执行每个命令时，更新状态。你可以把它写成一个单一的方法，或者做一些更花哨的事情，比如State的多态实现，每个状态都知道如何处理一些命令（例如LoginState需要USER和PASS，然后你把状态变成一个新的AuthorizedState)。

3) 我不记得在每个通道中使用 NIO 和许多异步写入器，但文档说它是线程安全的（我不会详细说明，因为我没有证据）。关于 OP_WRITE，请注意它会在写入缓冲区未满时发出信号。换句话说，正如here 所说：OP_WRITE 几乎总是准备就绪，即除非套接字发送缓冲区已满，因此您只会导致您的Selector.select() 方法无意识地旋转。

4) 是的。 Selector.select() 执行 blocking selection operation。

5) 我认为最困难的部分是从每客户端线程架构切换到读取和写入与处理分离的不同设计。完成此操作后，使用通道比使用自己的方式阻塞流更容易。

【讨论】：

你能增加一些精度吗？ Q1：我需要手动在缓冲区中使用get/put，还是在包装的字节数组中手动​​搜索\n，来分割行，或者有没有更好的方法？ Q2：我可以使用 SelectionKey 的用户对象来存储状态信息吗？ +为了更准确，我将在问题 1 中添加一个附加子问题。感谢您到目前为止的回答。