zoukankan      html  css  js  c++  java
  • IO模型之NIO代码及其实践详解

    一、简介

      NIO我们一般认为是New I/O(也是官方的叫法),因为它是相对于老的I/O类库新增的( JDK 1.4中的java.nio.*包中引入新的Java I/O库)。但现在都称之为Non-blocking I/O,即非阻塞I/O,因为这样叫,更能体现它的特点。而下文中的NIO,不是指整个新的I/O库,而是非阻塞I/O。

      NIO提供了与传统BIO模型中的Socket和ServerSocket相对应的SocketChannel和ServerSocketChannel两种不同的套接字通道实现。

      新增的着两种通道都支持阻塞和非阻塞两种模式。

      阻塞模式使用就像传统中的支持一样,比较简单,但是性能和可靠性都不好;非阻塞模式正好与之相反。

      对于低负载、低并发的应用程序,可以使用同步阻塞I/O来提升开发速率和更好的维护性;对于高负载、高并发的(网络)应用,应使用NIO的非阻塞模式来开发。

    二、NIO实现基础

      NIO对应linux IO 模型的IO多路复用模型,netty、tomcat(tomcat 6及以后版本,tomcat 6之前是基于BIO)的实现也是基于NIO。

                 

      I/O复用模型,是同步非阻塞,这里的非阻塞是指I/O读写,对应的是recvfrom操作,因为数据报文已经准备好,无需阻塞。说它是同步,是因为,这个执行是在一个线程里面执行的。有时候,还会说它又是阻塞的,实际上是指阻塞在select上面,必须等到读就绪、写就绪等网络事件。有时候我们又说I/O复用是多路复用,这里的多路是指N个连接,每一个连接对应一个channel,或者说多路就是多个channel。复用,是指多个连接复用了一个线程或者少量线程(在Tomcat中是Math.min(2,Runtime.getRuntime().availableProcessors()))。

      NIO的注册、轮询等待、读写操作协作关系如下图:

              

      1、多路复用器 Selector

      Selector的英文含义是“选择器”,Selector是Java  NIO 编程的基础,也可以称为为“轮询代理器”、“事件订阅器”、“channel容器管理机”都行。

      Selector中也会维护一个“已经注册的Channel”的容器(select使用数组,poll使用链表,epoll是红黑树+双向链表,因为JDK使用了epoll()代替传统的select实现,所以没有最大连接句柄1024/2048(32*32/32*64)的限制,具体原理请转到7层网络以及5种Linux IO模型以及相应IO基础文章末尾部分),应用程序将向Selector对象注册需要它关注的Channel,以及具体的某一个Channel会对哪些IO事件感兴趣。

      Selector提供选择已经就绪的任务的能力:Selector会不断轮询注册在其上的Channel,如果某个Channel上面发生读或者写事件,这个Channel就处于就绪状态,会被Selector轮询出来,然后通过SelectionKey可以获取就绪Channel的集合,进行后续的I/O操作。所以,只需要一个线程负责Selector的轮询,就可以接入成千上万的客户端。

      2、通道 Channel

      通道表示打开到 IO 设备(例如:文件、套接字,所以Channel有这两种实现:SelectableChannel 用户网络读写;FileChannel 用于文件操作)的连接。

            

      其中SelectableChannel有以下几种实现:

    • 所有被Selector(选择器)注册的通道,只能是继承了SelectableChannel类的子类。
    • ServerSocketChannel:应用服务器程序的监听通道。只有通过这个通道,应用程序才能向操作系统注册支持“多路复用IO”的端口监听。同时支持UDP协议和TCP协议。
    • ScoketChannel:TCP Socket套接字的监听通道,一个Socket套接字对应了一个客户端IP:端口 到 服务器IP:端口的通信连接。
    • DatagramChannel:UDP 数据报文的监听通道。

      Channel相比IO中的Stream流更加高效(底层的操作系统的通道一般都是全双工的,可以异步双向传输,所以全双工的Channel比流能更好的映射底层操作系统的API),但是必须和Buffer一起使用(若需要使用 NIO 系统,需要获取用于连接 IO 设备的Channel通道以及用于容纳数据的缓冲区Buffer。然后操作缓冲区,对数据进行处理。即通道中的数据总是要先读到一个Buffer,或者总是要从一个Buffer中写入)。也可以通过Channel通道向操作系统写数据。

      3、缓冲区Buffer

      Buffer是Channel操作读写的组件,包含一些要写入或者读出的数据。在NIO库中,所有数据都是用缓冲区处理的。在读取数据时,它是直接读到缓冲区中的;在写入数据时,也是写入到缓冲区中。任何时候访问NIO中的数据,都是通过缓冲区进行操作。缓冲区实际上是一个数组,并提供了对数据结构化访问以及维护读写位置等信息。

        具体的缓存区有这些:ByteBuffe、CharBuffer、 ShortBuffer、IntBuffer、LongBuffer、FloatBuffer、DoubleBuffer。他们实现了相同的接口:Buffer。但是在我们网络传输中都是使用byte数据类型。如图:

          

      Buffer中有三个重要参数:

      位置(position):当前缓冲区(Buffer)的位置,将从该位置往后读或写数据。

      容量(capacity):缓冲区的总容量上限。

      上限(limit):缓冲区的实际容量大小。

      写模式下Position为写入多少数据的位数标识,Limit等于Capacity;读模式下从Position读到limit,Position为0。写读模式切换使用flip()方法。

      flip方法将Buffer从写模式切换到读模式。调用flip()方法会将position设回0,并将limit设置成之前position的值。

    public Buffer flip() {
        limit = position;
        position = 0;
        mark = -1;
        return this;
    }

              

    三、代码实现

      1、服务端代码实现

    public class MultiplexerNioServer implements Runnable {
    
        private ServerSocketChannel serverSocketChannel;
        private Selector selector;
        private volatile boolean stop = false;
    
        /**
         * 初始化多路复用器 绑定监听端口
         *
         * @param port
         */
        public MultiplexerNioServer(int port) {
            try {
                serverSocketChannel = ServerSocketChannel.open();//获得一个serverChannel
                selector = Selector.open();////创建选择器  获得一个多路复用器
                serverSocketChannel.configureBlocking(false);//设置为非阻塞模式 如果为 true,则此通道将被置于阻塞模式;如果为 false,则此通道将被置于非阻塞模式
                serverSocketChannel.socket().bind(new InetSocketAddress(port), 1024);//绑定一个端口和等待队列长度
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);//把selector注册到channel,关注链接事件
            } catch (IOException e) {
                e.printStackTrace();
                System.exit(1);
            }
        }
    
        public void stop() {
            this.stop = true;    // 优雅停机
        }
    
        public void run() {
            while (!stop) {
                try {
                    //无论是否有读写事件发生,selector每隔1s被唤醒一次。如果一定时间内没有事件,就需要做些其他的事情,就可以使用带超时的
                    int client = selector.select(1000);
                    System.out.println("1:"+client);
                    // 阻塞,只有当至少一个注册的事件发生的时候才会继续.
              // int client = selector.select(); 不设置超时时间为线程阻塞,但是IO上支持多个文件描述符就绪
                    if (client == 0) {
                        continue;
                    }
                    System.out.println("2:"+client);
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> it = selectionKeys.iterator();
                    SelectionKey key = null;
                    while (it.hasNext()) {
                        key = it.next();
                        it.remove();
                        try {
                            //处理事件
                            handle(key);
                        } catch (Exception e) {
                            if (key != null) {
                                key.cancel();
                                if (key.channel() != null) {
                                    key.channel().close();
                                }
                            }
                        }
                    }
                } catch (Throwable e) {
                    e.printStackTrace();
                }finally {
    
                }
            }
    
            if (selector != null) {
                // selector关闭后会自动释放里面管理的资源
                try {
                    selector.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        public void handle(SelectionKey key) throws IOException {
            if (key.isValid()) {
                //连接事件
                if (key.isAcceptable()) {
                    ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                    // 通过ServerSocketChannel的accept创建SocketChannel实例
                    // 完成该操作意味着完成TCP三次握手,TCP物理链路正式建立
                    SocketChannel sc = ssc.accept();//3次握手
                    sc.configureBlocking(false);
                    sc.register(selector, SelectionKey.OP_READ);//连接建立后关注读事件
                }
    
                //读事件
                if (key.isReadable()) {
                    SocketChannel socketChannel = (SocketChannel) key.channel();
                    ByteBuffer readbuffer = ByteBuffer.allocate(1024);//写 0 1024  1024
    //                ByteBuffer readbuffer = ByteBuffer.allocateDirect(1024); //申请直接内存,也就是堆外内存
                    // 读取请求码流,返回读取到的字节数
                    int readBytes = socketChannel.read(readbuffer);
                    // 读取到字节,对字节进行编解码
                    if (readBytes > 0) {
                        // 将缓冲区当前的limit设置为position=0,用于后续对缓冲区的读取操作
                        readbuffer.flip();//读写模式反转
                        // 将缓冲区可读字节数组复制到新建的数组中
                        byte[] bytes = new byte[readbuffer.remaining()];
                        readbuffer.get(bytes);
                        String body = new String(bytes, "UTF-8");
                        System.out.println("input is:" + body);
    
                        res(socketChannel, body);
                    }else if(readBytes < 0){
                        // 链路已经关闭 释放资源
                        key.cancel();
                        socketChannel.close();
                    }else{
                        // 没有读到字节忽略
                    }
                }
    
            }
        }
    
        private void res(SocketChannel channel, String response) throws IOException {
            if (response != null && response.length() > 0) {
                byte[] bytes = response.getBytes();
                ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
                writeBuffer.put(bytes);
                writeBuffer.flip();
                channel.write(writeBuffer);
                System.out.println("res end");
            }
        }
    }

      可以看到,创建NIO服务端的主要步骤如下:

        1、打开ServerSocketChannel,监听客户端连接;

        2、绑定监听端口,设置连接为非阻塞模式;

        3、创建Reactor线程,创建多路复用器并启动线程。

        4、将ServerSocketChannel注册到Reactor线程中的Selector上,监听ACCEPT事件

        5、Selector轮询准备就绪的key

        6、Selector监听到新的客户端接入,处理新的接入请求,完成TCP三次握手,简历物理链路

        7、设置客户端链路为非阻塞模式

        8、将新接入的客户端连接注册到Reactor线程的Selector上,监听读操作,读取客户端发送的网络消息

        9、异步读取客户端消息到缓冲区

        10、对Buffer编解码,处理半包消息,将解码成功的消息封装成Task

        11、将应答消息编码为Buffer,调用SocketChannel的write将消息异步发送给客户端

      2、服务端启动

    public class NioServer {
        public static void main(String[] args) {
            int port=8080;
            MultiplexerNioServer nioServer=new MultiplexerNioServer(port);
            new Thread(nioServer,"nioserver-001").start();
        }
    }

      启动服务端后,打印堆栈信息可以看到,会一直阻塞在 select() 操作,等待请求的到来:

            

      通过控制台手动建立连接,测试如下:

            

      3、客户端代码实现

    public class NioClientHandler implements Runnable {
        private String host;
        private int port;
        private Selector selector;
        private SocketChannel socketChannel;
        private volatile boolean stop;
    
        public NioClientHandler(String host, int port) {
            this.host = host;
            this.port = port;
            try {
                // 创建选择器
                selector = Selector.open();
                // 打开监听通道
                socketChannel = SocketChannel.open();
                // 如果为 true,则此通道将被置于阻塞模式;如果为 false,则此通道将被置于非阻塞模式
                socketChannel.configureBlocking(false); // 开启非阻塞模式
            } catch (IOException e) {
                e.printStackTrace();
                System.exit(1);
            }
        }
    
        public void run() {
            try {
                doConnect();
            } catch (IOException e) {
                e.printStackTrace();
                System.exit(1);
            }
            while (!stop) {
                try {
                    int wait=selector.select(1000);
                    if(wait==0){
                        continue;
                    }
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> it = selectionKeys.iterator();
                    SelectionKey key = null;
    
                    while (it.hasNext()) {
                        key = it.next();
                        it.remove();
                        try {
                            handle(key);
                        } catch (Exception e) {
                            if (key != null) {
                                key.cancel();
                                if (key.channel() != null) {
                                    key.channel().close();
                                }
                            }
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    System.exit(1);
                }
            }
            if (selector != null) {
                try {
                    selector.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        private void doConnect() throws IOException {
            if (socketChannel.connect(new InetSocketAddress(host, port))) {
                socketChannel.register(selector, SelectionKey.OP_READ);
                doWrite(socketChannel);
            }else{
                socketChannel.register(selector, SelectionKey.OP_CONNECT);
            }
        }
    
        private void handle(SelectionKey key) throws IOException {
            if (key.isValid()) {
                SocketChannel sc = (SocketChannel) key.channel();
                if (key.isConnectable()) {
                    if (sc.finishConnect()) {
                        sc.register(selector, SelectionKey.OP_READ);
                        doWrite(sc);
                    } else {
                        System.exit(1);
                    }
                }
                if (key.isReadable()) {
                    ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                    int readBytes = sc.read(readBuffer);
                    if (readBytes > 0) {
                        readBuffer.flip();
                        byte[] bytes=new byte[readBuffer.remaining()];
                        readBuffer.get(bytes);
                        String body=new String(bytes,"UTF-8");
                        System.out.println("res"+body);
                        this.stop=true;
                    }else if(readBytes<0){
                        key.cancel();
                        sc.close();
                    }
    
                }
            }
        }
        
        private void doWrite(SocketChannel sc) throws IOException {
            // 将消息编码为字节数组
            byte[] request = "0123456789".getBytes();
            // 根据数组容量创建ByteBuffer
            ByteBuffer writeBuffer = ByteBuffer.allocate(request.length);
            // 将字节数组复制到缓冲区
            writeBuffer.put(request);
            // flip读写切换操作
            writeBuffer.flip();
            sc.write(writeBuffer);
            if (!writeBuffer.hasRemaining()) {
                System.out.println("写入完成");
            }
        }
    }

      4、启动客户端

    public class NioClient {
        public static void main(String[] args) {
            new Thread(new NioClientHandler("localhost", 8080), "nioClient-001").start();
        }
    }

      启动客户端后,会向服务端写一些数据,然后再从服务端接收数据,如下图所示:

                 

       客户端获得服务端的回复后会自动断开同服务端的连接,客户端的一次请求就此完成。

    四、BIO、NIO、AIO对比

             

  • 相关阅读:
    mysql timestamp字段定义的
    mybatis的Selective接口和普通接口的区别
    intllij IDE 中git ignore 无法删除target目录下的文件
    maven的单元测试中没有
    java volatile关键字
    RestExpress response中addHeader 导致stackOverflow
    log4j配置后行号乱码显示为?问号
    软件研发人员的职业发展规划
    CPU与内存互联的架构演变
    windows系统安装
  • 原文地址:https://www.cnblogs.com/jing99/p/12000371.html
Copyright © 2011-2022 走看看