NIO(生活篇)

　　今晚是个下雨天，写完今天最后一行代码，小鲁班起身合上电脑，用滚烫的开水为自己泡制了一桶老坛酸菜牛肉面。这大概是苦逼程序猿给接下来继续奋战的自己最好的馈赠。年轻的程序猿更偏爱坐在窗前，在夜晚中静静的享受独特的泡面香味。。。

　　科班出身的小鲁班虽然写了N多复杂（CRUD）代码，但仍口味清淡，他们往往不加或少加料包，由泡面热腾腾的蒸汽熏蒸自己的脸频，润湿又干又涩的双眼，抚慰受伤的心灵。然后，看着外边依然还是熙熙攘攘的车流和不属于自己的任何一个亮灯的窗口，却思考着如何才能成为ー个名垂青史的程序猿。小鲁班不迷茫。。。

　　"我们一起学猫叫，一起喵喵喵"~~~小鲁班放在书桌上的大哥大手机突然响了，打破了小鲁班脑子里美好的yy

　　小鲁班心想：都这么晚了，谁TM还打电话过来，拿起电话一看，哦原来是他表哥鲁班大师

　　鲁班大师：小老弟。晚上好嘛！

　　小鲁班：嘤嘤嘤，原来是大表哥呀，能和你通话真让我难以置信呀。

　　鲁班大师：听说你今天早上请两老去馆子喝早茶去了呀，有钱人，看来混的很不错嘛。

　　小鲁班：哎，别提了，等了半天才通知有位置（接待阻塞），坐下之后又没人来负责写菜单（点餐阻塞），写完菜单又没有人负责上菜，我去~气死老子

　　鲁班大师：哈哈哈哈哈，这馆子的老板也太奥特曼(out)了，现在规模大点的饭馆都采用NIO（同步非阻塞IO）模式啦。

　　小鲁班：额？，NIO是什么鬼，这和饭馆有什么关系呢？

　　鲁班大师： emmmmm，故事得从一段很长很长的网络编程模式历史开始说起呢~

　　S1.传统的网络编程模式（单线程下的通信）

　　

　　在单线程模式下，IO操作没完成的时候，无法返回，造成服务器线程阻塞，其他客户端不能连上服务端。

　　在只有一个餐厅服务员的情况下，服务员接待了一位客人，客人到餐桌上坐下后，服务员等待客人点餐，此时又有一个客人来吃饭，但是已经没有服务员去接待了，因为这个服务员在等待第一个客人点餐，直到第一个客人点完餐后，服务员把菜单交给厨房，然后才能去接待第二个进来的客人。。。（这样的服务客人早就走了）

　　那么我们来看看如何改进

　　S2改良后网络编程模式（多线程）

　　在S1中我们发现了一些问题，当IO阻塞的时候，服务端无法接受请求，因此S2改用了多线程模式

　　

　　在多线程模式下，只要有客户端连进来，我们都会为之创建一个线程专门去处理客户端的IO操作。当完成之后，线程就会自动销毁。但是这样会带来一个问题，就是线程的频繁的创建和销毁非常消耗服务器的资源。

　　饭馆里的老板面对这种情况，只好继续请服务员去写菜单了，来一个客人，就请一个服务员去负责客人的单子，问题是请服务员非常消耗老板的money呀，而且当写完单子后又要计算工资，这个过程非常耗时间。

　　PS：这里插入一些概念方便后文理解

　　线程轮询：

　　只是将当前线程不停地执行循环体，不进行线程状态的改变，所以响应速度更快。但当线程数不停增加时，性能下降明显，因为每个线程都需要执行，占用CPU时间。

　　线程阻塞：
　　让线程进入阻塞状态进行等待，当获得相应的信号（唤醒，时间） 时，才可以进入线程的准备就绪状态，准备就绪状态的所有线程，通过竞争CPU资源，进入运行状态。 阻塞的线程不会占用 CPU 时间， 不会导致 CPU 占用率过高，但线程从阻塞状态进入就绪状态的切换时间要比轮询略慢，消耗性能也多。

　　S3继续改良后的网络编程模式（线程池）

　　S2我们发现了这样的问题就是线程的创建和销毁非常损耗系统的性能，因此我们想到JDBC中连接池的解决方案，同样的，这里我们可以创建线程池

　　

　　启动服务后，事先创建100个线程，当有客户端连进来的时候，不需要创建，就给他分配一个线程用于IO读写操作，当客户完成IO操作完成之后就归还线程到线程池中而不是销毁，这样做的好处就是解决了在运行的时候线程创建和销毁对系统资源的损耗。同时也暴露了一些问题，其一在高并发的情况下，线程池中的线程不够用了，此时会造成客户端等待阻塞（当然也可以继续创建线程来解决），其二高并发环境下由于普遍线程都存在读写阻塞，使得各个线程一起频繁的进行上下文的切换，消耗的大量的资源，而这些资源本来是用来处理业务用的，现在则用来切换线程，这就大大的降低了系统有效的资源利用率。同时每个客户端都为他开了一个线程，在很多时候，其实客户端并不进行IO操作，就没必要为他创建线程，因为系统的无IO操作线程数多了的话会也占用CPU资源的。

　　老板觉得一直请人不划算，干脆就请30个人是在餐厅一个角落待命，当有客人坐下来的时候，就分配一个服务员去点餐。但是当有31客人同时来的时候，假设30个服务员都在等待写单，那么第31个客人假如要点餐的时候就没人为他服务了，同时点完餐时候的，突然客人想加餐，此时每个服务员都想着去抢到这个客户，竞争过程消耗了时间，同时得知道刚刚的账单都点了些什么还要相互交接未完成的任务，就更浪费人力物力。

　　S4再次改良后的网络编程模式（NIO）（非阻塞的IO多路复用机制）

　　S3我们发现线程池不够用，以及高并发情况下普遍线程都存在读写阻塞问题，使得各个线程一起频繁的进行上下文的切换，消耗的大量的资源。主要原因都是

1. 线程的IO阻塞导致线程状态频繁的切换消耗系统资源
2. 无IO操作线程占用CUP资源导致的

　　因此

　　针对问题1我们可以通过建立无阻塞环境，这样就不会因为阻塞导致线程状态的切换。

　　针对问题2我们可以通过改变线程的创建时机，不是Socket刚刚连上来的时候创建线程，而是等待需要进行IO操作的时候再去创建线程，从而减少无关线程的创建。

　　

　　这张图对比上面的题我们发现多个三个陌生的面孔，下面介绍一下他们

　　Channel表示为一个已经建立好的支持I/O操作的实体（如文件和网络）的连接，在此连接上进行数据的读写操作，使用的是Buffer缓冲区来实现读写。

　　ServerSocketChannel------->open() 获得实例 ----------->register(selector,accept) 将通道管理器和该通道绑定，并为该通道注册事件。

　　通过socket.getChannel()的方法获得通道inChannel

　　通道的数据传输是这样的

　　将Buffer的数据读入通道

　　int bytesWritten =inChannel.write(buffer);

　　从Channel读取数据到Buffer

　　int bytesRead =inChannel.read(buffer);

　　Selector一个专门的选择器来同时对多个Socket通道进行监听（轮询或阻塞）,当其中的某些Socket通道上有它感兴趣的事件发生时，这些通道就会变为可用状态,当状态是IO状态的时候，就会为他分配一个线程处理业务，当不是IO状态的时候只会为他注册一个接收（一共4种接收，连接，读，写），不会分配线程，这样的话就保证了，系统中存在的线程都是用来处理业务的而不是用来等待的，这样就能够减少线程，也就减少了线程上下文的切换损耗资源。利用 Selector可使一个单独的线程管理多个 Channel。Selector（多路复用器） 是非阻塞 IO 的核心。

　　Selector----->open()获得实例------>select()监听动作（读还是写还是连接，相当于之前的accept()方法），通过源码发现SelectorProvider.provider().poll()依赖于操作系统创建

　　Buffer缓冲区，就像一个数组，可以保存多个相同类型的数据（ByteBuffer,CharBuffer,.....DoubleBuffer）通过这个方法获取static XxxBuffer allocate(int capacity) 

其中里边有些方法例如clear、flip、rewind都是操作limit和position的值来实现重复读写，这样的话IO就不会阻塞，不会出现客户端在写入的时候，服务端不能写出造成线程的阻塞。简而言之，Channel负责传输，Buffer负责存储

　　position(初始的位置，读的时候，位置会移动)

　　limit(当你读取完成了，数据需要进行固定flip(),limit=position)

　　capacity(数组大小的一个容量)

　　clear()把position回归到原位

 　　

　　其实这里的Selector相当于一个接待主管，当有一个客人从大门（Channel）进来来吃饭的时候，先带它到位置上，给他安排一个台号，然后一直监听客人的需求，当客人需要点餐的时候，此时接待主管监听到了，就立马给他分配一个服务员去帮你它完成点餐，当客人需要加餐的时候，接待主管分配服务员到指定台号，然后只需要在账单（Buffer）上添加即可!

　　

　　小鲁班：哇塞，有点晕，但是我还是能看懂的，这些都是概念，表哥有代码么？

　　鲁班大师：代码嘛，我今天打排位的时候用鲁班被喷没皮肤，咳咳！

　　小鲁班：皮肤好说好说！

　　1.OIO服务端代码

public class OioServer {

    @SuppressWarnings("resource")
    public static void main(String[] args) throws Exception {

        ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();
        //创建socket服务,监听10101端口
        ServerSocket server=new ServerSocket(10101);
        System.out.println("服务器启动！");
        while(true){
            //获取一个套接字（阻塞）
            final Socket socket = server.accept();
            System.out.println("来个一个新客户端！");
            newCachedThreadPool.execute(new Runnable() {
                
                @Override
                public void run() {
                    //业务处理
                    handler(socket);
                }
            });
            
        }
    }
    
    /**
     * 读取数据
     * @param socket
     * @throws Exception
     */
    public static void handler(Socket socket){
            try {
                byte[] bytes = new byte[1024];
                InputStream inputStream = socket.getInputStream();
                
                while(true){
                    //读取数据（阻塞）
                    int read = inputStream.read(bytes);
                    if(read != -1){
                        System.out.println(new String(bytes, 0, read));
                    }else{
                        break;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }finally{
                try {
                    System.out.println("socket关闭");
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
    }
}

　　2.NIO服务端代码

public class NIOServer {
    // 通道管理器
    private Selector selector;

    /**
     * 获得一个ServerSocket通道，并对该通道做一些初始化的工作
     * 
     * @param port
     *            绑定的端口号
     * @throws IOException
     */
    public void initServer(int port) throws IOException {
        // 获得一个ServerSocket通道
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        // 设置通道为非阻塞
        serverChannel.configureBlocking(false);
        // 将该通道对应的ServerSocket绑定到port端口
        serverChannel.socket().bind(new InetSocketAddress(port));
        // 获得一个通道管理器
        this.selector = Selector.open();
        // 将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_ACCEPT事件,注册该事件后，
        // 当该事件到达时，selector.select()会返回，如果该事件没到达selector.select()会一直阻塞。
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
    }

    /**
     * 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理
     * 
     * @throws IOException
     */
    public void listen() throws IOException {
        System.out.println("服务端启动成功！");
        // 轮询访问selector
        while (true) {
            // 当注册的事件到达时，方法返回；否则,该方法会一直阻塞
            selector.select();
            // 获得selector中选中的项的迭代器，选中的项为注册的事件
            Iterator<?> ite = this.selector.selectedKeys().iterator();
            while (ite.hasNext()) {
                SelectionKey key = (SelectionKey) ite.next();
                // 删除已选的key,以防重复处理
                ite.remove();

                handler(key);
            }
        }
    }

    /**
     * 处理请求
     * 
     * @param key
     * @throws IOException
     */
    public void handler(SelectionKey key) throws IOException {
        
        // 客户端请求连接事件
        if (key.isAcceptable()) {
            handlerAccept(key);
            // 获得了可读的事件
        } else if (key.isReadable()) {
            handelerRead(key);
        }
    }

    /**
     * 处理连接请求
     * 
     * @param key
     * @throws IOException
     */
    public void handlerAccept(SelectionKey key) throws IOException {
        ServerSocketChannel server = (ServerSocketChannel) key.channel();
        // 获得和客户端连接的通道
        SocketChannel channel = server.accept();
        // 设置成非阻塞
        channel.configureBlocking(false);

        // 在这里可以给客户端发送信息哦
        System.out.println("新的客户端连接");
        // 在和客户端连接成功之后，为了可以接收到客户端的信息，需要给通道设置读的权限。
        channel.register(this.selector, SelectionKey.OP_READ);
    }

    /**
     * 处理读的事件
     * 
     * @param key
     * @throws IOException
     */
    public void handelerRead(SelectionKey key) throws IOException {
        // 服务器可读取消息:得到事件发生的Socket通道
        SocketChannel channel = (SocketChannel) key.channel();
        // 创建读取的缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        int read = channel.read(buffer);
        if(read > 0){
            byte[] data = buffer.array();
            String msg = new String(data).trim();
            System.out.println("服务端收到信息：" + msg);
            
            //回写数据
            ByteBuffer outBuffer = ByteBuffer.wrap("好的".getBytes());
            channel.write(outBuffer);// 将消息回送给客户端
        }else{
            System.out.println("客户端关闭");
            key.cancel();
        }
    }

    /**
     * 启动服务端测试
     * 
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
        NIOServer server = new NIOServer();
        server.initServer(8000);
        server.listen();
    }

}

 　　鲁班大师：好好理解，有什么不懂的就留言，你表哥先撤了，皮肤可别忘了送，不早了晚安~

　　小鲁班：好滴！

 　　

　　拓展一波，关于REDIS单线程为什么这么快？

　　其实这里用到的核心依然是NIO，我们来看一下REDIS的工作流程

　　

　　解析一下，IO多路复用器复制监听Socker连接的请求，把准备好的客户端请求压到一个队列中，这样避免了IO阻塞的等待，同时该线程只负责把请求压到队列中，而文件时间派发器则是从队列中获取请求，并处理请求，这里的处理方式是单线程的，避免了多线程频繁的上下文切换带来的资源耗费，同时该操作是存内存的操作，非常快

 　　总结有三点

       1）纯内存操作

       2）核心是基于非阻塞的IO多路复用机制

       3）单线程反而避免了多线程的频繁上下文切换问题