Tio Boot Docs

Appearance

AIO HTTP 服务器 ​

创建简单的 HTTP 服务器 ​

概述 ​

AIO（Asynchronous I/O，异步 I/O）是一种能够在处理网络请求时提升效率的方式，它能够非阻塞地处理多个请求。与传统的阻塞 I/O 不同，AIO 通过异步事件通知机制减少了线程的等待时间。在 Java 中，可以使用 AsynchronousServerSocketChannel 和 AsynchronousSocketChannel 来构建基于 AIO 的 HTTP 服务器。

本文介绍了如何使用 AIO 创建一个简单的 HTTP 服务器，该服务器能够接收客户端发送的请求并响应相应的内容。

代码实现 ​

java
package nexus.io.ip;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

public class AioHttpServerJava8 {

  public static void main(String[] args) {
    try {
      // 创建异步服务器通道并绑定到端口8080
      AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open();
      serverChannel.bind(new InetSocketAddress(8080));

      System.out.println("AIO HTTP Server started on port 8080...");

      // 使用非阻塞模式接受连接
      serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {
        @Override
        public void completed(AsynchronousSocketChannel clientChannel, Void attachment) {
          // 再次调用 accept()，以继续接受新连接
          serverChannel.accept(null, this);

          if (clientChannel != null && clientChannel.isOpen()) {
            handleRequest(clientChannel);
          }

        }

        @Override
        public void failed(Throwable exc, Void attachment) {
          serverChannel.accept(null, this);
          exc.printStackTrace();
        }
      });

      // 防止主线程退出
      Thread.currentThread().join();

    } catch (IOException | InterruptedException e) {
      e.printStackTrace();
    }
  }

  // 处理客户端连接
  private static void handleRequest(AsynchronousSocketChannel clientChannel) {
    ByteBuffer buffer = ByteBuffer.allocate(4096); // 增加 buffer 大小以处理较大请求

    clientChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
      @Override
      public void completed(Integer result, ByteBuffer attachment) {
        attachment.flip();
        String request = StandardCharsets.UTF_8.decode(attachment).toString();

        // 解析请求路径
        String requestPath = getRequestPath(request);
        if ("/ip".equals(requestPath)) {
          ipHandler(clientChannel, request);
          // 发送响应
          return;
        }

        // 其他路径,返回404
        writeHttpResponse(clientChannel, 404, "text/plain", "404 Not Found");

      }

      @Override
      public void failed(Throwable exc, ByteBuffer attachment) {
        exc.printStackTrace();
      }
    });
  }

  private static void ipHandler(AsynchronousSocketChannel clientChannel, String request) {
    // 解析请求，获取IP参数
    Map<String, String> requestMap = getRequestMap(request);
    String ip = requestMap.get("ip");

    String body = ip;

    // 构建HTTP响应
    writeHttpResponse(clientChannel, 200, "text/plain;charset=utf-8", body);
  }

  // 从请求中提取请求路径
  private static String getRequestPath(String request) {
    String[] lines = request.split("\r\n");
    for (String line : lines) {
      if (line.startsWith("GET") || line.startsWith("POST")) {
        String[] parts = line.split(" ");
        if (parts.length > 1) {
          String query = parts[1];
          return query.split("\\?")[0]; // 提取请求路径
        }
      }
    }
    return "/";
  }

  // 从请求中提取参数并封装为Map的方法
  private static Map<String, String> getRequestMap(String request) {
    Map<String, String> paramMap = new HashMap<>();
    String[] lines = request.split("\r\n");
    for (String line : lines) {
      if (line.startsWith("GET") || line.startsWith("POST")) {
        String[] parts = line.split(" ");
        if (parts.length > 1) {
          String query = parts[1];
          if (query.contains("?")) {
            String[] queryParams = query.substring(query.indexOf("?") + 1).split("&");
            for (String param : queryParams) {
              String[] keyValue = param.split("=");
              if (keyValue.length > 1) {
                paramMap.put(keyValue[0], keyValue[1]);
              } else {
                paramMap.put(keyValue[0], ""); // 如果没有值，则设为空字符串
              }
            }
          }
        }
      }
    }
    return paramMap;
  }

  private static void writeHttpResponse(AsynchronousSocketChannel clientChannel, int statusCode, String contentType, String body) {

    String statusMessage;
    switch (statusCode) {
    case 200:
      statusMessage = "OK";
      break;
    case 404:
      statusMessage = "Not Found";
      break;
    case 500:
      statusMessage = "Internal Server Error";
      break;
    default:
      statusMessage = "Unknown";
    }

    String response;
    String string = "HTTP/1.1 " + statusCode + " " + statusMessage + "\r\n" + "Content-Type: " + contentType + "\r\n";

    if (body != null) {
      byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
      response = string + "Content-Length: " + bytes.length + "\r\n" + "\r\n" + body;
    } else {
      response = string + "Content-Length: 0" + "\r\n" + "\r\n";
    }

    ByteBuffer responseBuffer = ByteBuffer.wrap(response.getBytes());
    clientChannel.write(responseBuffer, responseBuffer, new CompletionHandler<Integer, ByteBuffer>() {
      @Override
      public void completed(Integer result, ByteBuffer buffer) {
        try {
          clientChannel.close();
        } catch (IOException e) {
          e.printStackTrace();
        }
      }

      @Override
      public void failed(Throwable exc, ByteBuffer buffer) {
        exc.printStackTrace();
      }
    });
  }
}

请求示例 ​

http://localhost:8080/ip?ip=192.168.1.100

响应 ​

192.168.1.100

日志示例 ​

AIO HTTP Server started on port 8080...
Received request:
GET /ip?ip=192.168.1.100 HTTP/1.1
Host: localhost:8080
Connection: keep-alive
Pragma: no-cache
Cache-Control: no-cache
sec-ch-ua: "Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: "Windows"
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
Sec-Fetch-Site: none
Sec-Fetch-Mode: navigate
Sec-Fetch-User: ?1
Sec-Fetch-Dest: document
Accept-Encoding: gzip, deflate, br, zstd
Accept-Language: en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7
Cookie: SECKEY_ABVK=ycY75NLzlyo0bY0Kn7ZG5bVPp/l7na1YeWo9DWQ68ds%3D; BMAP_SECKEY=ycY75NLzlyo0bY0Kn7ZG5Z81bbL49sME2ZtV4ChhP7dmYvS5bFstYwymM9_vhT07nWUjJzQ0XX7FwomWPJJKjGukIC9JPxJKEKt9wn8aMUKBAY1tMttDz5xTxCST_QIROkSLyIiDMcL0dMLVxbcA4I1jdJfL_axpsM5vS_A-DhULyyNfAe6L08l_OBQ1rpK3; PHPSE
Extracted IP parameter: 192.168.1.100

代码讲解 ​

AIO 的运行原理 ​

在这段代码中，AIO 通过异步通道 AsynchronousServerSocketChannel 和 AsynchronousSocketChannel 实现非阻塞的网络通信。

  1. 创建服务器通道：服务器通过 AsynchronousServerSocketChannel 打开一个异步通道，并绑定到指定端口（8080）。

    java
    AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open();
serverChannel.bind(new InetSocketAddress(8080));

  2. 接受连接：服务器在 while(true) 循环中调用 accept() 方法，该方法返回一个 Future 对象，表示一个即将建立的连接。调用 get() 方法会等待并获取实际的客户端连接。

    java
    Future<AsynchronousSocketChannel> acceptFuture = serverChannel.accept();
AsynchronousSocketChannel clientChannel = acceptFuture.get();

  3. 处理请求：连接建立后，服务器通过 read() 方法读取客户端发送的数据。read() 方法是异步的，它接受一个缓冲区和 CompletionHandler，读取完成后会调用 completed() 方法。服务器通过解析请求中的数据路径来判断响应的内容。

    java
    clientChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
  @Override
  public void completed(Integer result, ByteBuffer attachment) {
    // 处理请求
  }
});

  4. 发送响应：通过 write() 方法将响应写入客户端，并使用 CompletionHandler 异步地完成响应的发送。当写入完成时调用 completed() 方法关闭客户端连接。

    java
    clientChannel.write(responseBuffer, responseBuffer, new CompletionHandler<Integer, ByteBuffer>() {
  @Override
  public void completed(Integer result, ByteBuffer buffer) {
    clientChannel.close();
  }
});

部分代码解释 ​

为什么要再次调用 AsynchronousServerSocketChannel 的 acccept 方法 ​

在异步 I/O 编程中，尤其是使用 AsynchronousServerSocketChannel 时，accept() 方法只会接受一次连接。这意味着每次有新的客户端连接时，都需要再次调用 accept() 方法来等待新的连接请求。如果不再次调用 accept()，服务器将不会继续接收新的客户端连接。

让我们详细解释一下：

1. accept() 方法的工作原理 ​

AsynchronousServerSocketChannel.accept() 是一个异步方法，用于等待客户端连接。当有客户端连接时，它会触发指定的回调（在这里是 CompletionHandler）。但该方法只会处理一个连接请求，处理完当前连接后它就“停止工作”了。

如果希望服务器能够继续接受后续的客户端连接，必须再次调用 accept() 方法。

2. 为什么要在回调中再次调用 accept()？ ​

当一个连接请求完成后，比如有客户端连接到服务器，accept() 的回调方法（即 CompletionHandler 的 completed() 方法）会被调用。在处理完这个客户端连接后，需要让服务器准备好接受下一个连接，这就是为什么需要在 completed() 回调中再次调用 accept() 方法的原因：

java
serverChannel.accept(null, this);

这个调用是为了：

  • 重新开始等待新的客户端连接：在处理完一个连接后，服务器需要继续监听端口，等待下一个连接请求。如果不这样做，服务器将无法接受更多的客户端连接。
  • 保持服务器能够处理多个客户端：每次一个新的客户端连接时，accept() 被调用以准备下一个连接请求，从而使服务器能不断处理多个客户端连接。
3. 异步模式的特性 ​

在异步 I/O 模式下，操作是非阻塞的，服务器不会停留在 accept() 调用处等待，而是会立即返回。只有当连接请求到达时，回调函数才会被调用。这种非阻塞行为使得我们可以并行处理多个客户端。

通过在 accept() 的回调中再次调用 accept()，确保了服务器能够继续接受新连接，并且不会阻塞在某个连接上。

总结： ​

accept() 方法只接受一个连接，处理完这个连接后，必须再次调用 accept() 来接受下一个连接。这样可以保证服务器能够持续监听并处理后续的客户端连接，而不会因为只处理一个连接后就停止工作。

为什么在 completed 的第一行调用 accept 更好？ ​

在 completed 方法的第一行就调用 serverChannel.accept(null, this);，这是一个常见的做法，确保在处理当前客户端连接时，服务器已经开始接受下一个客户端连接请求。

  1. 不阻塞新连接：在异步 I/O 模式下，处理一个连接的同时，希望服务器能立即开始接受新的连接请求。如果在 completed 的后面调用 accept，服务器在处理当前连接期间不会接受新的连接，可能导致其他客户端在等待。将 accept() 放在第一行意味着立即开始准备接受下一个连接，确保并发处理多个客户端。

  2. 保持高并发性能：在高并发场景下，服务器能够并行处理多个连接请求非常重要。如果 accept() 在连接处理结束后才被调用，那么服务器会浪费时间等待，无法在处理一个连接的同时接受新的连接。

将 serverChannel.accept(null, this); 放在 completed() 方法的第一行是一个最佳实践，确保服务器始终处于可接受新连接的状态，保持异步服务器的高并发性能和响应能力。

serverChannel 瞬时只能接受一个连接吗？ ​

是的，serverChannel.accept() 一次只能处理一个连接请求。这并不意味着服务器只能同时处理一个连接，而是说：

  • 每次调用 accept()：只能接受一个新的客户端连接。
  • 处理当前连接后：需要再次调用 accept() 来接受下一个客户端连接。

在异步 I/O 模式下，这个过程是非阻塞的。也就是说，可以同时接受并处理多个连接，但每个新的连接请求都需要通过 accept() 方法来启动。

异步 I/O 的并行处理机制

尽管 accept() 一次只能接受一个连接，但在异步 I/O 模式下，服务器在处理一个客户端连接时，仍然可以处理其他的连接。这是通过以下机制实现的：

  1. accept() 只负责接受连接：一旦连接成功建立，服务器会将连接交给 AsynchronousSocketChannel 来处理数据读写。

  2. 并行处理多个连接：每个客户端连接都有自己独立的 AsynchronousSocketChannel，服务器可以并行处理多个客户端的请求和响应。通过 read()、write() 等方法，服务器能够同时处理多个连接的数据流。

  3. 重复调用 accept()：每次成功接受连接后，需要再次调用 accept() 来准备接受下一个客户端的连接请求。正因为服务器是异步的，这样做不会阻塞正在进行的其他 I/O 操作。

如果出现异常，调用 failed() 方法，服务器会停止接受新的连接？ ​

当在异步 I/O 操作中出现异常时，CompletionHandler 的 failed() 方法会被调用。如果不在 failed() 方法中重新调用 accept()，那么服务器确实会停止接受新的连接。

为了确保服务器在发生异常时能够继续接受新的连接，需要在 failed() 方法中也调用 accept()。这样即使发生异常，服务器仍然能够处理新的连接请求。

buffer 变量的作用 ​

java
ByteBuffer buffer = ByteBuffer.allocate(4096); // 增加 buffer 大小以处理较大请求

在这段代码中，buffer 是用于存储从客户端读取的数据的缓存。当客户端发送请求时，数据会被读取到这个 ByteBuffer 中。提到的 "两个 buffer 变量" 指的是 buffer 既作为读操作的参数传递，也作为 CompletionHandler 的第二个泛型参数传递。下面解释一下原因：

buffer 的用途： ​

  1. 数据读取缓存：buffer 是实际用于存储客户端请求数据的缓存。当异步通道读取数据时，它需要一个 ByteBuffer 来存储这些数据。因此，在 clientChannel.read(buffer, buffer, handler) 中，第一个 buffer 是用于存储从客户端读取的数据。

  2. 异步操作的上下文：异步 I/O 操作中的 CompletionHandler 有两个泛型参数，第一个是操作结果（这里是读取的字节数 Integer），第二个是上下文对象（这里是 ByteBuffer）。在这段代码中，buffer 被作为上下文传入到 CompletionHandler，这样就可以在回调方法（completed）中继续使用同一个 ByteBuffer 来处理数据。因为这是异步操作，所以需要将 buffer 传入，以便在读取完成后可以继续处理它。

clientChannel.read 为什么要传入两个 ByteBuffer 对象 ​

buffer 变量确实只创建了一个 ByteBuffer 对象，并且在异步读取数据时两次传入同一个 buffer 作为参数。这里的 buffer 实际上起着不同的作用，因此两次传入的 buffer 是有必要的。让我详细解释一下：

1. 传入 buffer 到 read() 方法 ​

在代码中，这行是负责读取客户端数据的关键操作：

java
clientChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {...});
  • 第一个 buffer 是传递给 read() 方法的。AsynchronousSocketChannel.read() 方法的第一个参数是一个 ByteBuffer，用于存储从客户端读取的数据。也就是说，网络请求的数据会被写入这个 buffer 中。
2. 传入 buffer 作为 CompletionHandler 的上下文 ​

第二个 buffer 作为上下文传递给 CompletionHandler 的 attachment 参数。在异步操作完成时，CompletionHandler 会被调用。需要知道，在回调函数中如何访问到原始的 ByteBuffer 对象，这样可以处理客户端发送的数据。所以传入第二个 buffer 作为上下文，确保当 completed() 回调被触发时，能够继续访问到这个 buffer，从而读取和处理数据。

为什么需要两次传入 buffer？ ​

虽然传入的是同一个 ByteBuffer 对象，但它在 read() 方法和 CompletionHandler 中分别起着不同的作用：

  • 第一用途：作为 read() 方法的目标缓冲区，用于存储从客户端接收到的数据。
  • 第二用途：作为 CompletionHandler 的上下文，这样可以在异步读取完成后继续访问同一个 buffer。

如果只传递给 read()，而不作为 CompletionHandler 的上下文传入，异步回调完成时就无法轻松访问到这个缓冲区里的数据。而 CompletionHandler 的设计就是让可以携带上下文信息，比如 buffer，从而在回调时处理数据。

在异步编程中，将同一个 buffer 传递给 read() 和 CompletionHandler 是为了保持在异步操作完成后的上下文数据。即便是两次传入 buffer，这仍然是对同一个 ByteBuffer 对象的引用，确保数据的读取和处理能顺利完成。

请求处理过程 ​

  1. 一个请求发送到 AIO 时，AIO 为该请求分配的对象是 AsynchronousSocketChannel。

    当客户端发送请求时，AIO 服务器通过 AsynchronousServerSocketChannel.accept() 方法接受客户端连接，并为该连接分配一个 AsynchronousSocketChannel 对象。这个对象代表客户端和服务器之间的异步通信通道。AsynchronousSocketChannel 是用于处理单个请求的通道，每个客户端请求都会有一个独立的 AsynchronousSocketChannel 实例。

    java
    Future<AsynchronousSocketChannel> acceptFuture = serverChannel.accept();
AsynchronousSocketChannel clientChannel = acceptFuture.get();

  2. 后续处理是针对 AsynchronousSocketChannel 对象进行的镜像处理。

    这个对象被用于读取客户端请求数据和发送响应数据。具体步骤如下：

    • 读取请求数据：当客户端发送请求时，AsynchronousSocketChannel.read() 方法会启动异步读取操作。这个方法不会阻塞，它会将数据写入 ByteBuffer，并在读取完成后通过 CompletionHandler 回调函数处理数据。

      java
      clientChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
  @Override
  public void completed(Integer result, ByteBuffer attachment) {
    // 读取完成，处理请求
    String request = StandardCharsets.UTF_8.decode(attachment).toString();
  }

  @Override
  public void failed(Throwable exc, ByteBuffer attachment) {
    // 读取失败的处理
  }
});

    • 发送响应数据：处理完成后，服务器会根据请求内容构建响应并通过 AsynchronousSocketChannel.write() 方法将响应异步地发送给客户端。写入完成后，服务器会关闭该通道。

      java
      ByteBuffer responseBuffer = ByteBuffer.wrap(response.getBytes(StandardCharsets.UTF_8));
clientChannel.write(responseBuffer, responseBuffer, new CompletionHandler<Integer, ByteBuffer>() {
  @Override
  public void completed(Integer result, ByteBuffer buffer) {
    try {
      clientChannel.close(); // 关闭通道
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  @Override
  public void failed(Throwable exc, ByteBuffer buffer) {
    // 写入失败的处理
  }
});

默认线程池 ​

使用 AsynchronousServerSocketChannel.open() 而不指定任何参数时，Java NIO 使用了操作系统的本地异步 I/O 实现。在 Linux 上，这通常是通过 epoll 来实现的，非常高效，能够处理大量的并发连接。

  • 默认实现：使用 epoll，事件驱动，不需要为每个连接分配线程，能够高效地处理大量并发连接。

  • 自定义线程池的影响：一旦您为 AsynchronousChannelGroup 提供了自定义的线程池，即使是通过 withCachedThreadPool 方法，Java NIO 可能会切换到一种混合模式，部分使用线程池，部分使用事件驱动。这可能导致性能下降，因为线程池的配置和行为会影响到底层的异步 I/O 模型。

  • JVM 内部优化：Java NIO 的默认线程池（通过 AsynchronousChannelProvider.provider() 提供）由 JVM 自动管理，并且根据系统资源（如 CPU 核心数）进行了性能优化。不需要担心如何调优线程池的大小、队列长度等问题，这些细节由 JVM 内部处理。

-自动负载管理：默认线程池会根据系统的负载情况，自动调整线程的使用，确保异步 I/O 操作的高效执行，而不会轻易出现过载或资源耗尽的情况。

在您的 Java AIO HTTP 服务器代码中，AsynchronousServerSocketChannel 被用于处理异步的网络通信。针对您提出的两个问题，以下是详细的解答：

1. AsynchronousServerSocketChannel.open() 会导致线程被频繁创建和销毁吗？ ​

回答：不会。

AsynchronousServerSocketChannel.open() 方法本身并不会直接导致线程的频繁创建和销毁。相反，它依赖于底层的线程池来处理异步操作。具体来说：

  • 异步模型：AIO（Asynchronous I/O）采用的是事件驱动的异步模型，允许单个或少量的线程处理大量的并发连接。这意味着不需要为每个连接或请求创建独立的线程，从而避免了线程的频繁创建和销毁。
  • 线程池管理：Java 的异步通道（如AsynchronousServerSocketChannel）默认使用一个全局的AsynchronousChannelGroup，该组内部维护一个线程池。这些线程池会复用已有的线程来处理新的异步任务，而不是为每个任务都创建新的线程。

因此，AsynchronousServerSocketChannel 在处理并发连接时能够高效地利用线程资源，避免了线程的频繁创建和销毁，从而提升了性能和资源利用率。

2. AsynchronousServerSocketChannel 内部的线程池是什么？如何实现线程的复用？ ​

回答：

AsynchronousServerSocketChannel 依赖于 AsynchronousChannelGroup 来管理其内部的线程池。具体细节如下：

  • 默认线程池：

    • 当您调用 AsynchronousServerSocketChannel.open() 而不传递任何参数时，Java 会使用默认的 AsynchronousChannelGroup。这个默认的通道组通常基于全局的线程池实现。
    • 默认线程池的大小和行为依赖于 JVM 的实现，但通常是基于可用处理器数量动态调整的，能够高效地处理大量的异步任务。

  • 自定义线程池：

    • 如果您希望更细粒度地控制线程池（例如，线程数、线程工厂等），可以创建一个自定义的 AsynchronousChannelGroup 并将其传递给 AsynchronousServerSocketChannel.open(group) 方法。

    • 例如，使用 Executors.newFixedThreadPool 创建一个固定大小的线程池，然后基于该线程池创建 AsynchronousChannelGroup：

      java
      ExecutorService executor = Executors.newFixedThreadPool(10);
AsynchronousChannelGroup group = AsynchronousChannelGroup.withThreadPool(executor);
AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open(group);

  • 线程复用机制：

    • 线程池中的线程在完成一个异步任务后，不会被销毁，而是被标记为可用，并等待处理下一个任务。这种复用机制显著减少了线程的创建和销毁开销，提高了性能。
    • 线程池内部通常使用工作队列来管理待处理的任务，当有新的异步操作需要处理时，线程池会从队列中取出任务并分配给空闲的线程。

总结 ​

AsynchronousServerSocketChannel 通过依赖于高效的线程池机制，实现了线程的复用和资源的高效利用。这使得它非常适合用于构建高并发的网络服务器，而不会因为线程管理不善而导致性能瓶颈。

如果您对线程池的行为有特定的需求或优化目标，建议使用自定义的 AsynchronousChannelGroup 来精细控制线程池的配置。

附加参考资料：

多线程 Reactor 模型 ​

AsynchronousServerSocketChannel 使用的是单线程 Reactor 模型还是多线程 Reactor 模型？

答案：AsynchronousServerSocketChannel 使用的是多线程 Reactor 模型。

详细解释 ​

1. Reactor 模式概述 ​

Reactor 模式是一种事件驱动的设计模式，主要用于处理并发的输入事件。它通常包括以下组件：

  • Reactor：负责监听和分发事件。
  • Handlers：负责处理具体的事件。

Reactor 模式可以分为单线程和多线程两种实现：

  • 单线程 Reactor 模式：所有事件的监听和处理都在单一线程中完成，适用于连接数较少或处理逻辑简单的场景。
  • 多线程 Reactor 模式：Reactor 使用多个线程来处理事件，适用于高并发和复杂处理逻辑的场景。
2. AsynchronousServerSocketChannel 的内部实现 ​

AsynchronousServerSocketChannel 使用了 多线程 Reactor 模式，其内部机制包括以下几个方面：

  • AsynchronousChannelGroup：

    • AsynchronousServerSocketChannel 依赖于 AsynchronousChannelGroup 来管理其内部的线程池。
    • 默认情况下，如果不显式指定 AsynchronousChannelGroup，Java 会为异步通道创建一个全局的、基于固定大小线程池的 AsynchronousChannelGroup。
    • 这个线程池通常会根据系统的可用处理器数量自动调整线程数，以便高效地处理大量并发的异步任务。

  • 线程池的复用：

    • 线程池中的线程在完成一个异步任务后，并不会被销毁，而是被复用来处理下一个任务。
    • 这种复用机制显著减少了线程的创建和销毁开销，提高了系统的性能和资源利用率。

  • 事件处理的并行性：

    • 由于使用了多线程，多个事件（如新连接的接受、数据的读取/写入等）可以并行处理。
    • 这使得 AsynchronousServerSocketChannel 能够高效地处理高并发的连接和请求。

总结 ​

  • 请求到达时，AIO 会为每个请求分配一个 AsynchronousSocketChannel 对象，代表服务器与该客户端的通信通道。
  • 所有的读写操作都是通过这个通道对象进行的，读操作将数据写入缓冲区，写操作将响应发送回客户端。
  • 通道使用 CompletionHandler 进行异步操作，读写数据的处理都是通过回调完成的。

MIT Licensed

Copyright © 2023-present litongjava