NIO实现非阻塞式Socket通信

xiaoxiao2025-04-24  6

首先我们明确阻塞模型

•FileChannel   : 由上图可知道,该channel 只有阻塞模式。

•DatagramChannel

•SocketChannel

ServerSocketChannel

三个网络channel 可以通过configureBlocking 方法,设置非阻塞模式

关于阻塞与非阻塞,详情请看同步异步,阻塞非阻塞区别

关于nio 的非阻塞

NIO ,只有在 网络中,使用  Selector 实现多路复用,才是实现了非阻塞模式。

单纯的使用  Buffer channel 是无法实现非阻塞的。

非阻塞: 请求资源,若资源未准备好,则不等待,而是直接返回 一个 error标识。但是为了获取资源,我们就要不停的去请求,仍然不能实现现实的非阻塞。

所以使用多路复用,当前通过一个单独的线程使用 select 去监视多个channel , 那么只需要该线程不断的去请求,看哪些channel 是准备好的。

Selector 非阻塞通信核心

Selector负责监控所有已经注册的ChannelSelector通过SelectionKey来关联Channel

特别注意上边  ServerSocketChannel , 这个channel 只有一个,负责监听客户端来的连接。他监听一个端口。

将该 ServerSocketChannel  要设置为非阻塞。 因为其被绑定在 select 上,server.accept(),若是阻塞的,将会造成阻塞,不能去监听其他channel .所以 ServerSocketChannel 和SocketChannel 地位是平等的。

select 对  ServerSocketChannel  监听 连接请求。有连接是, 其去新建channel 绑定为读请求,并绑定到select.

详情看代码。

服务器开启时,只有 ServerSocketChannel   一个channel绑定在Select 上,当来请求时,创建新的channel,并绑定到Select上去监听读就绪。

群聊服务器nio非阻塞式 public class NServer { // 用于检测所有Channel状态的Selector private Selector selector = null; static final int PORT = 30000; // 定义实现编码、解码的字符集对象 private Charset charset = Charset.forName("UTF-8"); public void init()throws IOException { selector = Selector.open(); // 通过open方法来打开一个未绑定的ServerSocketChannel实例 ServerSocketChannel server = ServerSocketChannel.open(); InetSocketAddress isa = new InetSocketAddress("127.0.0.1", PORT); // 将该ServerSocketChannel绑定到指定IP地址 server.bind(isa); // 设置ServerSocket以非阻塞方式工作 server.configureBlocking(false); // 将server注册到指定Selector对象 server.register(selector, SelectionKey.OP_ACCEPT); while (selector.select() > 0) { // 依次处理selector上的每个已选择的SelectionKey for (SelectionKey sk : selector.selectedKeys()) { // 从selector上的已选择Key集中删除正在处理的SelectionKey selector.selectedKeys().remove(sk); // ① // 如果sk对应的Channel包含客户端的连接请求 if (sk.isAcceptable()) // ② { // 调用accept方法接受连接,产生服务器端的SocketChannel //此处Server就一个可以直接上边的获取到,可以不通过SelectionKey 获取 SocketChannel sc = server.accept(); // 设置采用非阻塞模式 sc.configureBlocking(false); // 将该SocketChannel也注册到selector sc.register(selector, SelectionKey.OP_READ); // 将sk对应的Channel设置成准备接受其他请求 sk.interestOps(SelectionKey.OP_ACCEPT); } // 如果sk对应的Channel有数据需要读取 if (sk.isReadable()) // ③ { // 获取该SelectionKey对应的Channel,该Channel中有可读的数据 SocketChannel sc = (SocketChannel)sk.channel(); // 定义准备执行读取数据的ByteBuffer ByteBuffer buff = ByteBuffer.allocate(1024); String content = ""; // 开始读取数据 try { while(sc.read(buff) > 0) { buff.flip(); content += charset.decode(buff); } // 打印从该sk对应的Channel里读取到的数据 System.out.println("读取的数据:" + content); // 将sk对应的Channel设置成准备下一次读取 sk.interestOps(SelectionKey.OP_READ); } // 如果捕捉到该sk对应的Channel出现了异常,即表明该Channel // 对应的Client出现了问题,所以从Selector中取消sk的注册 catch (IOException ex) { // 从Selector中删除指定的SelectionKey sk.cancel(); if (sk.channel() != null) { sk.channel().close(); } } // 如果content的长度大于0,即聊天信息不为空 if (content.length() > 0) { // 遍历该selector里注册的所有SelectionKey for (SelectionKey key : selector.keys()) { // 获取该key对应的Channel Channel targetChannel = key.channel(); // 如果该channel是SocketChannel对象 if (targetChannel instanceof SocketChannel) { // 将读到的内容写入该Channel中 SocketChannel dest = (SocketChannel)targetChannel; dest.write(charset.encode(content)); } } } } } } } public static void main(String[] args) throws IOException { new NServer().init(); } } 群聊客户端 public class NClient { // 定义检测SocketChannel的Selector对象 private Selector selector = null; static final int PORT = 30000; // 定义处理编码和解码的字符集 private Charset charset = Charset.forName("UTF-8"); // 客户端SocketChannel private SocketChannel sc = null; public void init()throws IOException { selector = Selector.open(); InetSocketAddress isa = new InetSocketAddress("127.0.0.1", PORT); // 调用open静态方法创建连接到指定主机的SocketChannel sc = SocketChannel.open(isa); // 设置该sc以非阻塞方式工作 sc.configureBlocking(false); // 将SocketChannel对象注册到指定Selector sc.register(selector, SelectionKey.OP_READ); // 启动读取服务器端数据的线程 new ClientThread().start(); // 创建键盘输入流 Scanner scan = new Scanner(System.in); while (scan.hasNextLine()) { // 读取键盘输入 String line = scan.nextLine(); // 将键盘输入的内容输出到SocketChannel中 sc.write(charset.encode(line)); } } // 定义读取服务器数据的线程 private class ClientThread extends Thread { public void run() { try { while (selector.select() > 0) // ① { // 遍历每个有可用IO操作Channel对应的SelectionKey for (SelectionKey sk : selector.selectedKeys()) { // 删除正在处理的SelectionKey selector.selectedKeys().remove(sk); // 如果该SelectionKey对应的Channel中有可读的数据 if (sk.isReadable()) { // 使用NIO读取Channel中的数据 SocketChannel sc = (SocketChannel)sk.channel(); ByteBuffer buff = ByteBuffer.allocate(1024); String content = ""; while(sc.read(buff) > 0) { buff.flip(); content += charset.decode(buff); } // 打印输出读取的内容 System.out.println("聊天信息:" + content); // 为下一次读取作准备 sk.interestOps(SelectionKey.OP_READ); } } } } catch (IOException ex) { ex.printStackTrace(); } } } public static void main(String[] args) throws IOException { new NClient().init(); } } Selector APISelectionKey  API
转载请注明原文地址: https://www.6miu.com/read-5029050.html

最新回复(0)