nio enhancement

hutool-socket/src/main/java/cn/hutool/socket/aio/AioServer.java

@@ -1,13 +1,5 @@
 package cn.hutool.socket.aio;
 
-import java.io.Closeable;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.SocketOption;
-import java.nio.ByteBuffer;
-import java.nio.channels.AsynchronousChannelGroup;
-import java.nio.channels.AsynchronousServerSocketChannel;
-
 import cn.hutool.core.io.IORuntimeException;
 import cn.hutool.core.io.IoUtil;
 import cn.hutool.core.thread.ThreadFactoryBuilder;
@@ -16,6 +8,14 @@ import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
 import cn.hutool.socket.SocketConfig;
 
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketOption;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousChannelGroup;
+import java.nio.channels.AsynchronousServerSocketChannel;
+
 /**
  * 基于AIO的Socket服务端实现
  *
@@ -76,11 +76,7 @@ public class AioServer implements Closeable {
 	 * @param sync 是否阻塞
 	 */
 	public void start(boolean sync) {
-		try {
-			doStart(sync);
-		} catch (IOException e) {
-			throw new IORuntimeException(e);
-		}
+		doStart(sync);
 	}
 
 	/**
@@ -173,9 +169,8 @@ public class AioServer implements Closeable {
 	 * 开始监听
 	 *
 	 * @param sync 是否阻塞
-	 * @throws IOException IO异常
 	 */
-	private void doStart(boolean sync) throws IOException {
+	private void doStart(boolean sync) {
 		log.debug("Aio Server started, waiting for accept.");
 
 		// 接收客户端连接

hutool-socket/src/main/java/cn/hutool/socket/nio/AcceptHandler.java

@@ -0,0 +1,38 @@
+package cn.hutool.socket.nio;
+
+import cn.hutool.core.io.IORuntimeException;
+import cn.hutool.log.StaticLog;
+
+import java.io.IOException;
+import java.nio.channels.CompletionHandler;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+
+/**
+ * 接入完成回调，单例使用
+ * 
+ * @author looly
+ */
+public class AcceptHandler implements CompletionHandler<ServerSocketChannel, NioServer> {
+
+	@Override
+	public void completed(ServerSocketChannel serverSocketChannel, NioServer nioServer) {
+		SocketChannel socketChannel;
+		try {
+			// 获取连接到此服务器的客户端通道
+			socketChannel = serverSocketChannel.accept();
+			StaticLog.debug("Client [{}] accepted.", socketChannel.getRemoteAddress());
+		} catch (IOException e) {
+			throw new IORuntimeException(e);
+		}
+
+		// SocketChannel通道的可读事件注册到Selector中
+		NioUtil.registerChannel(nioServer.getSelector(), socketChannel, Operation.READ);
+	}
+
+	@Override
+	public void failed(Throwable exc, NioServer nioServer) {
+		StaticLog.error(exc);
+	}
+
+}

hutool-socket/src/main/java/cn/hutool/socket/nio/ChannelHandler.java

@@ -0,0 +1,17 @@
+package cn.hutool.socket.nio;
+
+import java.nio.channels.SocketChannel;
+
+/**
+ * NIO数据处理接口，通过实现此接口，可以从{@link SocketChannel}中读写数据
+ *
+ */
+public interface ChannelHandler {
+
+	/**
+	 * 处理NIO数据
+	 *
+	 * @param socketChannel {@link SocketChannel}
+	 */
+	void handle(SocketChannel socketChannel);
+}

hutool-socket/src/main/java/cn/hutool/socket/nio/NioClient.java

@@ -2,7 +2,7 @@ package cn.hutool.socket.nio;
 
 import cn.hutool.core.io.IORuntimeException;
 import cn.hutool.core.io.IoUtil;
-import cn.hutool.core.thread.ThreadFactoryBuilder;
+import cn.hutool.core.thread.ThreadUtil;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -10,12 +10,8 @@ import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
-import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
 import java.util.Iterator;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
 
 /**
  * NIO客户端
@@ -24,159 +20,133 @@ import java.util.concurrent.ThreadFactory;
  * @since 4.4.5
  */
 public abstract class NioClient implements Closeable {
-
-    private Selector selector;
-    private SocketChannel channel;
-    private ExecutorService executorService;
-
-    /**
-     * 构造
-     *
-     * @param host 服务器地址
-     * @param port 端口
-     */
-    public NioClient(String host, int port) {
-        init(new InetSocketAddress(host, port));
-    }
-
-    /**
-     * 构造
-     *
-     * @param address 服务器地址
-     */
-    public NioClient(InetSocketAddress address) {
-        init(address);
-    }
-
-    /**
-     * 初始化
-     *
-     * @param address 地址和端口
-     * @return this
-     */
-    public NioClient init(InetSocketAddress address) {
-        try {
-            //创建一个SocketChannel对象，配置成非阻塞模式
-            this.channel = SocketChannel.open();
-            channel.configureBlocking(false);
-
-            //创建一个选择器，并把SocketChannel交给selector对象
-            this.selector = Selector.open();
-            channel.register(selector, SelectionKey.OP_CONNECT);
-
-            //发起建立连接的请求，这里会立即返回，当连接建立完成后，SocketChannel就会被选取出来
-            channel.connect(address);
-        } catch (IOException e) {
-            throw new IORuntimeException(e);
-        }
-        return this;
-    }
+	private Selector selector;
+	private SocketChannel channel;
 
 	/**
-	 * 检查连接是否建立完成
+	 * 构造
+	 *
+	 * @param host 服务器地址
+	 * @param port 端口
 	 */
-    public boolean waitConnect() throws IOException {
-    	boolean isConnect = false;
-		while (0 != this.selector.select()) {
+	public NioClient(String host, int port) {
+		init(new InetSocketAddress(host, port));
+	}
+
+	/**
+	 * 构造
+	 *
+	 * @param address 服务器地址
+	 */
+	public NioClient(InetSocketAddress address) {
+		init(address);
+	}
+
+	/**
+	 * 初始化
+	 *
+	 * @param address 地址和端口
+	 * @return this
+	 */
+	public NioClient init(InetSocketAddress address) {
+		try {
+			//创建一个SocketChannel对象，配置成非阻塞模式
+			this.channel = SocketChannel.open();
+			channel.configureBlocking(false);
+			channel.connect(address);
+
+			//创建一个选择器，并把SocketChannel交给selector对象
+			this.selector = Selector.open();
+			channel.register(this.selector, SelectionKey.OP_READ);
+
+			// 等待建立连接
+			//noinspection StatementWithEmptyBody
+			while (false == channel.finishConnect()){}
+		} catch (IOException e) {
+			throw new IORuntimeException(e);
+		}
+		return this;
+	}
+
+	/**
+	 * 开始监听
+	 */
+	public void listen() {
+		ThreadUtil.execute(() -> {
+			try {
+				doListen();
+			} catch (IOException e) {
+				e.printStackTrace();
+			}
+		});
+	}
+
+	/**
+	 * 开始监听
+	 *
+	 * @throws IOException IO异常
+	 */
+	private void doListen() throws IOException {
+		while (this.selector.isOpen() && 0 != this.selector.select()) {
+			// 返回已选择键的集合
 			final Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
 			while (keyIter.hasNext()) {
-				//连接建立完成
-				SelectionKey key = keyIter.next();
-				if (key.isConnectable()) {
-					if (this.channel.finishConnect()) {
-						this.channel.register(selector, SelectionKey.OP_READ);
-						isConnect = true;
-					}
-				}
+				handle(keyIter.next());
 				keyIter.remove();
-				break;
-			}
-			if (isConnect) {
-				break;
 			}
 		}
-		return isConnect;
 	}
 
-    /**
-     * 开始监听
-     */
-    public void listen() {
-		this.executorService = Executors.newSingleThreadExecutor(r -> {
-            final Thread thread = Executors.defaultThreadFactory().newThread(r);
-            thread.setName("nio-client-listen");
-            return thread;
-        });
-		this.executorService.execute(() -> {
-            try {
-                doListen();
-            } catch (IOException e) {
-                e.printStackTrace();
-            }
-        });
-    }
-
-    /**
-     * 开始监听
-     *
-     * @throws IOException IO异常
-     */
-    private void doListen() throws IOException {
-        while (0 != this.selector.select()) {
-            // 返回已选择键的集合
-            final Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
-            while (keyIter.hasNext()) {
-                handle(keyIter.next());
-                keyIter.remove();
-            }
-        }
-    }
-
-    /**
-     * 处理SelectionKey
-     *
-     * @param key SelectionKey
-     */
-    private void handle(SelectionKey key) throws IOException {
-        // 读事件就绪
-        if (key.isReadable()) {
-            final SocketChannel socketChannel = (SocketChannel) key.channel();
-            read(socketChannel);
-        }
-    }
-
-    /**
-     * 处理读事件<br>
-     * 当收到读取准备就绪的信号后，回调此方法，用户可读取从客户端传出来的消息
-     *
-     * @param socketChannel SocketChannel
-     */
-    protected abstract void read(SocketChannel socketChannel);
-
-    /**
-     * 实现写逻辑<br>
-     * 当收到写出准备就绪的信号后，回调此方法，用户可向客户端发送消息
-     *
-     * @param datas 发送的数据
-     * @return this
-     */
-    public NioClient write(ByteBuffer... datas) {
-        try {
-            this.channel.write(datas);
-        } catch (IOException e) {
-            throw new IORuntimeException(e);
-        }
-        return this;
-    }
-
-    public void closeListen() {
-		this.executorService.shutdown();
+	/**
+	 * 处理SelectionKey
+	 *
+	 * @param key SelectionKey
+	 */
+	private void handle(SelectionKey key) {
+		// 读事件就绪
+		if (key.isReadable()) {
+			final SocketChannel socketChannel = (SocketChannel) key.channel();
+			read(socketChannel);
+		}
 	}
 
-    @Override
-    public void close() {
-        IoUtil.close(this.selector);
-        IoUtil.close(this.channel);
-		closeListen();
-    }
+	/**
+	 * 处理读事件<br>
+	 * 当收到读取准备就绪的信号后，回调此方法，用户可读取从客户端传出来的消息
+	 *
+	 * @param socketChannel SocketChannel
+	 */
+	protected abstract void read(SocketChannel socketChannel);
+
+	/**
+	 * 实现写逻辑<br>
+	 * 当收到写出准备就绪的信号后，回调此方法，用户可向客户端发送消息
+	 *
+	 * @param datas 发送的数据
+	 * @return this
+	 */
+	public NioClient write(ByteBuffer... datas) {
+		try {
+			this.channel.write(datas);
+		} catch (IOException e) {
+			throw new IORuntimeException(e);
+		}
+		return this;
+	}
+
+	/**
+	 * 获取SocketChannel
+	 *
+	 * @return SocketChannel
+	 * @since 5.3.10
+	 */
+	public SocketChannel getChannel() {
+		return this.channel;
+	}
+
+	@Override
+	public void close() {
+		IoUtil.close(this.selector);
+		IoUtil.close(this.channel);
+	}
 }

hutool-socket/src/main/java/cn/hutool/socket/nio/NioServer.java

@@ -2,12 +2,11 @@ package cn.hutool.socket.nio;
 
 import cn.hutool.core.io.IORuntimeException;
 import cn.hutool.core.io.IoUtil;
+import cn.hutool.log.Log;
 
 import java.io.Closeable;
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.net.ServerSocket;
-import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.ServerSocketChannel;
@@ -20,10 +19,14 @@ import java.util.Iterator;
  * @author looly
  *
  */
-public abstract class NioServer implements Closeable {
+public class NioServer implements Closeable {
+	private static final Log log = Log.get();
+
+	private static final AcceptHandler ACCEPT_HANDLER = new AcceptHandler();
 
 	private Selector selector;
 	private ServerSocketChannel serverSocketChannel;
+	private ChannelHandler handler;
 
 	/**
 	 * 构造
@@ -45,23 +48,52 @@ public abstract class NioServer implements Closeable {
 			// 打开服务器套接字通道
 			this.serverSocketChannel = ServerSocketChannel.open();
 			// 设置为非阻塞状态
-			serverSocketChannel.configureBlocking(false);
-			// 获取通道相关联的套接字
-			final ServerSocket serverSocket = serverSocketChannel.socket();
+			this.serverSocketChannel.configureBlocking(false);
 			// 绑定端口号
-			serverSocket.bind(address);
+			this.serverSocketChannel.bind(address);
 
 			// 打开一个选择器
-			selector = Selector.open();
+			this.selector = Selector.open();
 			// 服务器套接字注册到Selector中 并指定Selector监控连接事件
-			serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
+			this.serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
 		} catch (IOException e) {
 			throw new IORuntimeException(e);
 		}
 
+		log.debug("Server listen on: [{}]...", address);
+
 		return this;
 	}
 
+	/**
+	 * 设置NIO数据处理器
+	 *
+	 * @param handler {@link ChannelHandler}
+	 * @return this
+	 */
+	public NioServer setChannelHandler(ChannelHandler handler){
+		this.handler = handler;
+		return this;
+	}
+
+	/**
+	 * 获取{@link Selector}
+	 *
+	 * @return {@link Selector}
+	 */
+	public Selector getSelector(){
+		return this.selector;
+	}
+
+	/**
+	 * 启动NIO服务端，即开始监听
+	 *
+	 * @see #listen()
+	 */
+	public void start(){
+		listen();
+	}
+
 	/**
 	 * 开始监听
 	 */
@@ -79,7 +111,7 @@ public abstract class NioServer implements Closeable {
 	 * @throws IOException IO异常
 	 */
 	private void doListen() throws IOException {
-		while (0 != this.selector.select()) {
+		while (this.selector.isOpen() && 0 != this.selector.select()) {
 			// 返回已选择键的集合
 			final Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
 			while (keyIter.hasNext()) {
@@ -97,35 +129,13 @@ public abstract class NioServer implements Closeable {
 	private void handle(SelectionKey key) {
 		// 有客户端接入此服务端
 		if (key.isAcceptable()) {
-			// 获取通道 转化为要处理的类型
-			final ServerSocketChannel server = (ServerSocketChannel) key.channel();
-			SocketChannel socketChannel;
-			try {
-				// 获取连接到此服务器的客户端通道
-				socketChannel = server.accept();
-			} catch (IOException e) {
-				throw new IORuntimeException(e);
-			}
-
-			// SocketChannel通道的可读事件注册到Selector中
-			registerChannel(selector, socketChannel, Operation.READ);
+			ACCEPT_HANDLER.completed((ServerSocketChannel) key.channel(), this);
 		}
 
 		// 读事件就绪
 		if (key.isReadable()) {
 			final SocketChannel socketChannel = (SocketChannel) key.channel();
-			read(socketChannel);
-
-			// SocketChannel通道的可写事件注册到Selector中
-			registerChannel(selector, socketChannel, Operation.WRITE);
-		}
-
-		// 写事件就绪
-		if (key.isWritable()) {
-			final SocketChannel socketChannel = (SocketChannel) key.channel();
-			write(socketChannel);
-			// SocketChannel通道的可读事件注册到Selector中
-			registerChannel(selector, socketChannel, Operation.READ);
+			handler.handle(socketChannel);
 		}
 	}
 
@@ -134,42 +144,4 @@ public abstract class NioServer implements Closeable {
 		IoUtil.close(this.selector);
 		IoUtil.close(this.serverSocketChannel);
 	}
-
-	/**
-	 * 处理读事件<br>
-	 * 当收到读取准备就绪的信号后，回调此方法，用户可读取从客户端传出来的消息
-	 * 
-	 * @param socketChannel SocketChannel
-	 */
-	protected abstract void read(SocketChannel socketChannel);
-
-	/**
-	 * 实现写逻辑<br>
-	 * 当收到写出准备就绪的信号后，回调此方法，用户可向客户端发送消息
-	 * 
-	 * @param socketChannel SocketChannel
-	 */
-	protected abstract void write(SocketChannel socketChannel);
-
-	/**
-	 * 注册通道到指定Selector上
-	 * 
-	 * @param selector Selector
-	 * @param channel 通道
-	 * @param ops 注册的通道监听类型
-	 */
-	private void registerChannel(Selector selector, SelectableChannel channel, Operation ops) {
-		if (channel == null) {
-			return;
-		}
-
-		try {
-			channel.configureBlocking(false);
-			// 注册通道
-			//noinspection MagicConstant
-			channel.register(selector, ops.getValue());
-		} catch (IOException e) {
-			throw new IORuntimeException(e);
-		}
-	}
 }

hutool-socket/src/main/java/cn/hutool/socket/nio/NioUtil.java

@@ -0,0 +1,37 @@
+package cn.hutool.socket.nio;
+
+import cn.hutool.core.io.IORuntimeException;
+
+import java.io.IOException;
+import java.nio.channels.SelectableChannel;
+import java.nio.channels.Selector;
+
+/**
+ * NIO工具类
+ *
+ * @since 5.4.0
+ */
+public class NioUtil {
+
+	/**
+	 * 注册通道的指定操作到指定Selector上
+	 *
+	 * @param selector Selector
+	 * @param channel 通道
+	 * @param ops 注册的通道监听（操作）类型
+	 */
+	public static void registerChannel(Selector selector, SelectableChannel channel, Operation ops) {
+		if (channel == null) {
+			return;
+		}
+
+		try {
+			channel.configureBlocking(false);
+			// 注册通道
+			//noinspection MagicConstant
+			channel.register(selector, ops.getValue());
+		} catch (IOException e) {
+			throw new IORuntimeException(e);
+		}
+	}
+}