NIO开发流程

╰+攻爆jí腚メ 2022-02-13 15:14 197阅读 0赞

# **[NIO原理剖析][NIO]** #

Java原生NIO开发大致流程:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70][]

### 第一步，绑定一个服务的端口 ###

这与传统阻塞IO中的ServerSocket类似，没什么好说的

### 第二步，打开通道管理器Selector并在Selector上注册一个事件 ###

当注册的事件发生时，Selector.select()会返回，否则一直阻塞。这一步很有意思，也是NIO第一个与传统IO不同的地方，NIO通过一个Selector线程可以管理大量客户端连接，反之传统IO一个客户端连接进来必须创建一个新的线程为它服务(当然你可以使用连接池)，我们知道线程对服务端来说是十分宝贵的资源，一个服务端进程所包含的线程是有限的；此外，每个线程会占用一定的内存空间，过多的线程可能导致内存溢出，这种情况下你可能会到想对虚拟机进行调优，比如通过修改参数-Xss限制单个线程大小，但这又可能导致StackOverFlow；另外，线程调度需要切换上下文，对于操作系统，它需要通过TCB(线程控制块)来对线程进行调度，过多的上下文切换浪费了CPU时间，降低了系统效率。

### 第三步，轮循访问Selector，当注册的事件到达时，方法返回 ###

下面的代码可以看到，方法整体是一个死循环，轮询访问Selector，发生某些已经注册在Selector上的事件时，该方法返回。可以通过selector.selectedKeys获取发生的事件，

该方法返回的是一个泛型集合Set<SelectionKey>，遍历这个集合与遍历普通集合没有什么不同，这里我们通过迭代器迭代的原因是我们需要删除已经处理的Key，避免重复处理：

public void listen() throws IOException {// 轮询访问selector
            while (true) {
                // 当注册的事件到达时，方法返回；否则,该方法会一直阻塞
                selector.select();
                // 获得selector中选中的项的迭代器，选中的项为注册的事件
                Iterator<?> ite = this.selector.selectedKeys().iterator();
                while (ite.hasNext()) {
                    SelectionKey key = (SelectionKey) ite.next();
                    //删除已选的key,以防重复处理
                    ite.remove();
    　　　　　　　　　//这里可以写我们自己的处理逻辑
                    handle(key);
                }
            }
    }

在第二步时，已经在Selector上注册了Accept事件，当这里的selector.select()返回时，代表客户端已经可以连接了，在handle方法里可以处理这个事件：

public void handle(SelectionKey key) throws IOException {
        // 客户端请求连接事件
        if (key.isAcceptable()){
          //从Key里可以很方便的取到注册这个事件的Channel
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            // 获得和客户端连接的通道
            SocketChannel channel = server.accept();
            // 设置成非阻塞
            channel.configureBlocking(false);
            logger.info("客户端已经连接！");
            //客户端连接在通道管理器Selector上注册读事件
            channel.register(this.selector, SelectionKey.OP_READ);
        }
    }

上面的代码很简单，我们通过key获取到注册它的那个Channel，在这里是ServerSocketChannel，通过server.accept()获取客户端连接，这里同样可以类比到传统的阻塞

IO，在阻塞IO中我们可以通过ServerSocket.accept获取到socket，唯一不同的是，阻塞IO中的accept方法是阻塞操作，而NIO中是非阻塞的。

当然，仅仅是连接到客户端并没有什么用处，服务端需要有读写数据的能力，比如你可以用NIO实现一个Http服务器(当然最佳实践使用Netty等框架)。所以我们需要在Selector

上注册读事件，同样，当读事件发生时，执行我们自己的业务逻辑。下面是修改后的代码：

public void handle(SelectionKey key) throws IOException {
        if (key.isAcceptable()){
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            SocketChannel channel = server.accept();
            channel.configureBlocking(false);
            logger.info("客户端已经连接！");
            channel.register(this.selector, SelectionKey.OP_READ);
        } else if(key.isReadable()){
            SocketChannel channel = (SocketChannel) key.channel();
            // 创建读取缓冲
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            //读取到Buffer中
            int read = channel.read(buffer);
            if(read > 0){
                byte[] data = buffer.array();
                String msg = new String(data).trim();
                logger.info("receive msg: {}",msg)
                
                //回写数据
                ByteBuffer outBuffer = ByteBuffer.wrap("OK".getBytes());
                channel.write(outBuffer);
            }else{
                logger.info("client closed!!!");
                key.cancel();
            }
        }
    }

### **完整代码：** ###

客户端代码：

import static cn.enjoyedu.ch01.Ch01Const.DEFAULT_PORT;
    import static cn.enjoyedu.ch01.Ch01Const.DEFAULT_SERVER_IP;
    public class NioClient {
    
        private static NioClientHandle nioClientHandle;
    
        public static void start(){
            if(nioClientHandle !=null)
                nioClientHandle.stop();
            nioClientHandle = new NioClientHandle(DEFAULT_SERVER_IP,DEFAULT_PORT);
            new Thread(nioClientHandle,"Client").start();
        }
        //向服务器发送消息
        public static boolean sendMsg(String msg) throws Exception{
            nioClientHandle.sendMsg(msg);
            return true;
        }
        public static void main(String[] args) throws Exception {
            start();
            Scanner scanner = new Scanner(System.in);
            while(NioClient.sendMsg(scanner.next()));
        }
    }

import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Set;
    
    public class NioClientHandle implements Runnable{
        private String host;
        private int port;
        private Selector selector;
        private SocketChannel socketChannel;
    
        private volatile boolean started;
    
        public NioClientHandle(String ip, int port) {
            this.host = ip;
            this.port = port;
            try {
                //创建选择器
                selector = Selector.open();
                //打开通道
                socketChannel = SocketChannel.open();
                //如果为 true，则此通道将被置于阻塞模式；
                // 如果为 false，则此通道将被置于非阻塞模式
                socketChannel.configureBlocking(false);
                started = true;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        public void stop(){
            started = false;
        }
    
        @Override
        public void run() {
            try {
                doConnect();
            } catch (IOException e) {
                e.printStackTrace();
                System.exit(1);
            }
            //循环遍历selector
            while(started){
                try {
                    //阻塞,只有当至少一个注册的事件发生的时候才会继续
                    selector.select();
                    //获取当前有哪些事件可以使用
                    Set<SelectionKey> keys = selector.selectedKeys();
                    //转换为迭代器
                    Iterator<SelectionKey> it = keys.iterator();
                    SelectionKey key = null;
                    while(it.hasNext()){
                        key = it.next();
                        it.remove();
                        try {
                            handleInput(key);
                        } catch (IOException e) {
                            e.printStackTrace();
                            if(key!=null){
                                key.cancel();
                                if(key.channel()!=null){
                                    key.channel().close();
                                }
                            }
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
    
            //selector关闭后会自动释放里面管理的资源
            if(selector!=null){
                try {
                    selector.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        //具体的事件处理方法
        private void handleInput(SelectionKey key) throws IOException{
            if(key.isValid()){
                //获得关心当前事件的channel
                SocketChannel sc = (SocketChannel)key.channel();
                if(key.isConnectable()){//连接事件
                    if(sc.finishConnect()){}
                    else{System.exit(1);}
                }
                //有数据可读事件
                if(key.isReadable()){
                    //创建ByteBuffer，并开辟一个1M的缓冲区
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    //读取请求码流，返回读取到的字节数
                    int readBytes = sc.read(buffer);
                    //读取到字节，对字节进行编解码
                    if(readBytes>0){
                        //将缓冲区当前的limit设置为position,position=0，
                        // 用于后续对缓冲区的读取操作
                        buffer.flip();
                        //根据缓冲区可读字节数创建字节数组
                        byte[] bytes = new byte[buffer.remaining()];
                        //将缓冲区可读字节数组复制到新建的数组中
                        buffer.get(bytes);
                        String result = new String(bytes,"UTF-8");
                        System.out.println("accept message:"+result);
                    }else if(readBytes<0){
                        key.cancel();
                        sc.close();
                    }
                }
            }
        }
    
        //发送消息
        private void doWrite(SocketChannel channel,String request)
                throws IOException {
            //将消息编码为字节数组
            byte[] bytes = request.getBytes();
            //根据数组容量创建ByteBuffer
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            //将字节数组复制到缓冲区
            writeBuffer.put(bytes);
            //flip操作
            writeBuffer.flip();
            //发送缓冲区的字节数组
            channel.write(writeBuffer);
        }
        private void doConnect() throws IOException {
            /*如果此通道处于非阻塞模式，
            则调用此方法将启动非阻塞连接操作。
            如果立即建立连接，就像本地连接可能发生的那样，则此方法返回true。
            否则，此方法返回false，
            稍后必须通过调用finishConnect方法完成连接操作。*/
            if(socketChannel.connect(new InetSocketAddress(host,port))){}
            else{
                //连接还未完成，所以注册连接就绪事件，向selector表示关注这个事件
                socketChannel.register(selector,SelectionKey.OP_CONNECT);
            }
        }
        //写数据对外暴露的API
        public void sendMsg(String msg) throws Exception{
            socketChannel.register(selector,SelectionKey.OP_READ);
            doWrite(socketChannel,msg);
        }
    }

服务端代码：

import static cn.enjoyedu.ch01.Ch01Const.DEFAULT_PORT;
    
    public class NioServer {
    
        private static NioServerHandle nioServerHandle;
    
        public static void start(){
            if(nioServerHandle !=null)
                nioServerHandle.stop();
            nioServerHandle = new NioServerHandle(DEFAULT_PORT);
            new Thread(nioServerHandle,"Server").start();
        }
        public static void main(String[] args){
            start();
        }
    }

import java.io.IOException;
    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.Set;
    
    import static cn.enjoyedu.ch01.Ch01Const.response;
    
    public class NioServerHandle implements Runnable{
        private Selector selector;
        private ServerSocketChannel serverChannel;
        private volatile boolean started;
    
        public NioServerHandle(int port) {
    
            try {
                selector = Selector.open();
                serverChannel = ServerSocketChannel.open();
                serverChannel.configureBlocking(false);
                serverChannel.socket().bind(new InetSocketAddress(port));
                serverChannel.register(selector,SelectionKey.OP_ACCEPT);
                started = true;
                System.out.println("服务器已启动，端口号："+port);
            } catch (IOException e) {
                e.printStackTrace();
            }
    
        }
        public void stop(){
            started = false;
        }
        @Override
        public void run() {
            //循环遍历selector
            while(started){
                try{
                    //阻塞,只有当至少一个注册的事件发生的时候才会继续.
    				selector.select();
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> it = keys.iterator();
                    SelectionKey key = null;
                    while(it.hasNext()){
                        key = it.next();
                        it.remove();
                        try{
                            handleInput(key);
                        }catch(Exception e){
                            if(key != null){
                                key.cancel();
                                if(key.channel() != null){
                                    key.channel().close();
                                }
                            }
                        }
                    }
                }catch(Throwable t){
                    t.printStackTrace();
                }
            }
            //selector关闭后会自动释放里面管理的资源
            if(selector != null)
                try{
                    selector.close();
                }catch (Exception e) {
                    e.printStackTrace();
                }
        }
        private void handleInput(SelectionKey key) throws IOException{
            if(key.isValid()){
                //处理新接入的请求消息
                if(key.isAcceptable()){
                    //获得关心当前事件的channel
                    ServerSocketChannel ssc = (ServerSocketChannel)key.channel();
                    //通过ServerSocketChannel的accept创建SocketChannel实例
                    //完成该操作意味着完成TCP三次握手，TCP物理链路正式建立
                    SocketChannel sc = ssc.accept();
                    System.out.println("======socket channel 建立连接" );
                    //设置为非阻塞的
                    sc.configureBlocking(false);
                    //连接已经完成了，可以开始关心读事件了
                    sc.register(selector,SelectionKey.OP_READ);
                }
                //读消息
                if(key.isReadable()){
                    System.out.println("======socket channel 数据准备完成，" +
                            "可以去读==读取=======");
                    SocketChannel sc = (SocketChannel) key.channel();
                    //创建ByteBuffer，并开辟一个1M的缓冲区
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    //读取请求码流，返回读取到的字节数
                    int readBytes = sc.read(buffer);
                    //读取到字节，对字节进行编解码
                    if(readBytes>0){
                        //将缓冲区当前的limit设置为position=0，
                        // 用于后续对缓冲区的读取操作
                        buffer.flip();
                        //根据缓冲区可读字节数创建字节数组
                        byte[] bytes = new byte[buffer.remaining()];
                        //将缓冲区可读字节数组复制到新建的数组中
                        buffer.get(bytes);
                        String message = new String(bytes,"UTF-8");
                        System.out.println("服务器收到消息：" + message);
                        //处理数据
                        String result = response(message) ;
                        //发送应答消息
                        doWrite(sc,result);
                    }
                    //链路已经关闭，释放资源
                    else if(readBytes<0){
                        key.cancel();
                        sc.close();
                    }
                }
    
            }
        }
        //发送应答消息
        private void doWrite(SocketChannel channel,String response)
                throws IOException {
            //将消息编码为字节数组
            byte[] bytes = response.getBytes();
            //根据数组容量创建ByteBuffer
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            //将字节数组复制到缓冲区
            writeBuffer.put(bytes);
            //flip操作
            writeBuffer.flip();
            //发送缓冲区的字节数组
            channel.write(writeBuffer);
        }
    }

## 总结 ##

本文大致讲述了使用NIO进行服务器端开发的大致流程，但代码显然仍然存在问题，其一是我们只使用了一个线程执行所有操作，包括接收客户端连接，读取数据，返回数据，对于这个简单的Demo来说已经足够了，但在实际的服务器开发中，例如你想使用NIO开发自己的HTTP服务器，服务器本地需要做大量操作，包括解析用户请求，根据请求路由到某一个Action执行业务逻辑，这其中又很可能某些数据从数据库读取，渲染模板等操作，十分耗时，这无疑又称为系统的瓶颈，再者，使用单一线程不能充分利用多核CPU提供的计算能力。下一篇中会看到，在基于Reactor模型的Netty中，会使用一个Boss线程接收客户端请求，使用多个Worker线程执行具体的业务逻辑。[原文连接][NIO]

[NIO]: https://www.cnblogs.com/showing/p/6759653.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70]: /images/20220213/7c617328bbfa44c2be8749513ea98d54.png