Hutool AIO 封装 - AioServer 和 AioClient

背景

在 JDK7 及以上版本中，引入了异步 I/O（AIO）的 Socket 库。Hutool 对其进行了一定封装，提供了更简洁的 API 进行异步通信开发。AioServer 和 AioClient 是 Hutool 中用于处理异步通信的核心类。

• Hutool 提供的封装极大简化了 Java 原生 AIO 的使用，尤其是在 I/O 处理逻辑上，直接通过 SimpleIoAction 定义行为，减少了复杂性。
• 在实际开发中，AioServer 和 AioClient 适合用于处理轻量级、并发性要求较高的网络应用。

1. AioServer 用法详解

示例代码：

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.BufferUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.log.StaticLog;
import cn.hutool.socket.aio.AioServer;
import cn.hutool.socket.aio.AioSession;
import cn.hutool.socket.aio.SimpleIoAction;
import java.nio.ByteBuffer;

public class HutoolAioServerDemo {
    public static void main(String[] args) {
        // 创建一个 AIO 服务端，监听端口为 8899
        AioServer aioServer = new AioServer(8899);
        
        // 设置自定义的 I/O 处理逻辑
        aioServer.setIoAction(new SimpleIoAction() {
            
            // 当有客户端连接时触发
            @Override
            public void accept(AioSession session) {
                StaticLog.debug("【客户端】：{} 连接。", session.getRemoteAddress());
                
                // 发送欢迎信息给客户端
                session.write(BufferUtil.createUtf8("=== Welcome to Hutool socket server. ==="));
            }
            
            // 当有数据读取到时触发
            @Override
            public void doAction(AioSession session, ByteBuffer data) {
                // 打印接收到的数据
                Console.log(data);

                // 如果数据已读完，构造并发送 HTTP 响应
                if(!data.hasRemaining()) {
                    StringBuilder response = StrUtil.builder()
                            .append("HTTP/1.1 200 OK\r\n")
                            .append("Date: ").append(DateUtil.formatHttpDate(DateUtil.date())).append("\r\n")
                            .append("Content-Type: text/html; charset=UTF-8\r\n")
                            .append("\r\n")
                            .append("Hello Hutool socket");
                    
                    session.writeAndClose(BufferUtil.createUtf8(response));
                } else {
                    // 继续读取数据
                    session.read();
                }
            }
        });

        // 启动服务器，阻塞运行
        aioServer.start(true);
    }
}

代码说明：

1. AioServer(int port): 构造一个 AioServer，监听指定端口。
2. setIoAction(SimpleIoAction action): 设置 I/O 处理逻辑，定义了如何处理客户端的连接和数据读写。
3. accept(AioSession session): 当客户端连接时触发，可用于进行握手或发送欢迎消息等。
4. doAction(AioSession session, ByteBuffer data): 当有数据到达时触发，用于解析和处理客户端发送的数据。
5. start(boolean block): 启动服务器并开始监听端口。block 参数决定是否阻塞运行。

实际开发中的使用场景：

• 轻量级 HTTP 服务器：可以通过这种方式快速搭建简单的 HTTP 服务。
• 自定义通信协议服务端：开发自定义协议服务，比如聊天服务器、推送服务器等。

2. AioClient 用法详解

示例代码：

import cn.hutool.core.thread.ThreadFactoryBuilder;
import cn.hutool.core.util.StrUtil;
import cn.hutool.socket.aio.AioClient;
import cn.hutool.socket.aio.AioSession;
import cn.hutool.socket.aio.SimpleIoAction;
import cn.hutool.core.io.IoUtil;
import cn.hutool.core.io.BufferUtil;
import cn.hutool.core.exceptions.SocketRuntimeException;
import cn.hutool.core.exceptions.IORuntimeException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.concurrent.ExecutionException;

public class HutoolAioClientDemo {
    public static void main(String[] args) {
        // 创建一个固定线程池的异步通道组
        final AsynchronousChannelGroup group;
        try {
            group = AsynchronousChannelGroup.withFixedThreadPool(
                    Runtime.getRuntime().availableProcessors(), // 根据CPU核心数设置线程池大小
                    ThreadFactoryBuilder.create().setNamePrefix("Hutool-socket-").build() // 设置线程名称前缀
            );
        } catch (IOException e) {
            throw new IORuntimeException(e);
        }

        // 创建客户端通道
        AsynchronousSocketChannel channel;
        try {
            channel = AsynchronousSocketChannel.open(group);
        } catch (IOException e) {
            throw new IORuntimeException(e);
        }

        // 尝试连接到服务端
        try {
            channel.connect(new InetSocketAddress("localhost", 8899)).get();
        } catch (InterruptedException | ExecutionException e) {
            IoUtil.close(channel);
            throw new SocketRuntimeException(e);
        }

        // 创建 AioClient，设置 I/O 处理逻辑
        AioClient client = new AioClient(channel, new SimpleIoAction() {
            @Override
            public void doAction(AioSession session, ByteBuffer data) {
                if(data.hasRemaining()) {
                    Console.log(StrUtil.utf8Str(data));
                    session.read(); // 继续读取数据
                }
                Console.log("数据处理完毕");
            }
        });

        // 发送数据到服务端
        client.write(ByteBuffer.wrap("Hello".getBytes()));
        // 开始读取服务端的响应数据
        client.read();

        // 关闭客户端
        client.close();
    }
}

代码说明：

1. AsynchronousChannelGroup: 维护一个线程池用于管理客户端的异步操作，建议全局持有单例。
2. AsynchronousSocketChannel.open(group): 打开客户端通道并关联到指定的线程组。
3. connect(InetSocketAddress address): 尝试连接到指定的地址（如本地服务）。
4. AioClient(channel, SimpleIoAction action): 创建 AioClient 并设置 I/O 处理逻辑。
5. write(ByteBuffer buffer): 发送数据到服务端。
6. read(): 读取服务端的响应数据。
7. close(): 关闭客户端并释放资源。

实际开发中的使用场景：

• 客户端工具：与 AioServer 通信的客户端工具，处理各种请求和响应。
• 数据推送客户端：通过异步方式与服务器建立长连接，处理数据推送和实时更新。

注意事项：

1. AsynchronousChannelGroup 是一个资源密集型对象，建议作为单例全局持有。
2. AIO 模型适用于高并发场景，如即时通讯、推送服务等。