学习NIO 之 使用方法

 学习NIO 之 使用方法

 一丶什么是NIO

 　　NIO是非阻塞IO, 也可以说是新IO.在读写时并不会阻塞.

二丶同步与异步, 阻塞与非阻塞

 　　同步, 一个任务的完成依赖于另一个任务的完成, 需要等待另一个任务的完成, 才能执行本任务.

　　异步, 一个任务的完成不依赖另一个任务的完成,无须等待.

　　阻塞, 是相对于CPU来说的,  挂起当前线程.

　　非阻塞, 无须挂起当前线程, 可以去执行其他程序.

三丶与IO的对比

　　a. 旧IO是阻塞的, 新IO是非阻塞的.

　　b. 旧IO是面向流的, 只能单向读写, 新IO是面向缓冲的, 可以双向读写

　　在使用旧IO做Socket连接时, 由于单向读写, 当没有数据时, 会挂起当前线程, 阻塞等待, 为防止影响其它连接, 需要为每个连接新建线程处理. 然而系统资源是有限的, 不能过多的新建线程. 线程过多带来线程上下文的切换, 从来带来更大的性能损耗. 因此需要使用新IO进行IO多路复用, 使用一个线程来监听所有Socket连接, 使用本线程或者其他线程处理连接

四丶UNIX 5种IO模型

　　5种IO模型均可分为两个阶段, 一个是等待数据到达内核缓冲区, 一个是将数据从内核空间拷贝到用户空间, 其中前4种为同步IO, 最后1种为异步IO, 前4种的第二阶段完全相同

 　  a. I/O阻塞

　　　　阻塞等待数据到达内核空间, 最后将数据拷贝到用户空间

　　b. I/O非阻塞

　　　　轮训检查数据是否到达内核空间, 数据就绪后将数据拷贝到用户空间

　　c. I/O多路复用

　　IO多路复用, 主要思想是一个线程阻塞监听所有IO事件, 当事件到达时, 分发给对应处理线程, 将数据拷贝到用户空间进行处理

　　IO多路复用又分为3种实现方式 -- select/poll/epoll,

　　select方式会将需要监听的文件描述符从用户空间拷贝到内核空间, 内核会通过轮询该文件描述符数组,来判断对应事件是否就绪, 当事件就绪时, 就会将会整个文件描述符数组拷贝到用户空间. 所以select存在几个缺点: 1) 由于采用了数组保存文件描述符, 所以一般最大限制数量为1024   2)两次文件描述符的拷贝  3) 用户空间和内核空间上下文切换

　　poll方式与select方式类似, 只不过没有了最大文件描述符数量的限制, 因为采用了链表保存.

　　epoll方式则不同, 采用了红黑树保存需要监听的文件描述符, 并为该文件描述符注册回调函数, 当事件到达时, 会调用回调函数, 将红黑树中对应的文件描述符, 保存到一个双向链表中. 最后通过判断该双向链表是否为空即可.

　　三者的更详细区别可以查看此博文

　　d. 信号驱动I/O

　　　　在文件描述符就绪时发送SIGIO信号通知

　　e. 异步I/O

　　　　两个阶段都不用等待, 由内核完成, 数据到达内核缓冲区之后, 内核将数据拷贝到用户空间, 完成之后再通知用户程序.

　　

五丶Buffer的介绍

 　　buffer是缓冲区， 本质与其他缓冲(如BufferInputStream)一样， 里面都是字节数组(byte[])， 它在于控制一次性从数据源(磁盘等)读取多个字节，减少与数据源的交互次数，从而提高性能。

　　 a. buffer有几个重要属性：
        　　capacity: 缓冲区数组的长度
        　　position: 下一个要操作元素的位置
        　　limit: 下一个不可操作元素的位置
        　　mark: 用于记录当前position的前一个位置或者默认是0

　　 b. ByteBuffer几个重要方法

    // 创建缓冲区
      ByteBuffer allocate(int capacity);  //在堆中创建缓冲区
      ByteBuffer allocateDirect(int capacity); //在堆外内存中创建缓冲区
      
      // 存数据进buffer, position位置会随着数据的增加而变化
      ByteBuffer put(byte b);
      ByteBuffer put(byte[] src);
      
      // 读取buffer中的缓冲数据
      byte get();  //读取position对应的位置的数据， 并"移动"position (+1)
      
      // 将position设置为0， 一般用于读操作之后的写操作
      Buffer flip();
      
      // 判断是否还有元素可操作，(position<limit)
      boolean hasRemaining();
      
      // 清空数据
      Buffer clear();
      
      // 使用mark标记position的位置
      Buffer mark();
      
      // 重置position, 将position变为之前mark的操作
      Buffer reset();

 　　c. 注意事项
        在调用#put(byte[])方法将数据存进buffer中后， 需要调用#flip()方法， 将position指针移到开头， 然后才能将数据写进其他地方， 因为position表示下一个要操作的元素的位置, 操作示意图可以查看此博文

六丶Channel的介绍

 　　channel是一个通道，与io中的stream类似， 都是用于读取输出数据， 不同之处在于， channel是双向的， stream是单向的， channel是不能直接操作数据，需要将数据读取到缓冲区buffer之中，然后缓冲区中读取数据，或者将数据写入缓冲区buffer中，然后在将buffer中的数据写入通道中

七丶Selector的介绍

　　选择器，用于注册各种channel感兴趣的事件, 共有4种感兴趣的事件： OP_READ, OP_WRITE, OP_CONNECT, OP_ACCEPT

   ServerSocketChannel ssc=ServerSocketChannel.open();
   ssc.configureBlocking(false); // 设置非阻塞io
   //绑定端口 backlog设为1024
   ssc.socket().bind(new InetSocketAddress(8040),1024000000);
   ssc.register(selector, SelectionKey.OP_ACCEPT);

   SocketChannel sc=ssc.accept();
   sc.configureBlocking(false);
   sc.register(selector, SelectionKey.OP_READ);

   selector.select(); 是一个阻塞方法， 当没有感兴趣的事件时，会阻塞等待。
   selector.selectedKeys(); 返回感兴趣的所有事

　　   SelectionKey#isAcceptable(), #isReadable(), #isWritable(), #isConnectable() 判断发生的是哪种事件。

八丶完整示例代码

public class NioServerTests {

    public static void main(String[] args) throws IOException {

        // channel , selector
        ByteBuffer buffer=ByteBuffer.allocate(5);
        Selector selector=Selector.open();


//        Path path=FileSystems.getDefault().getPath("test.txt");
//        Channel channel=FileChannel.open(path, StandardOpenOption.WRITE);

        ServerSocketChannel ssc=ServerSocketChannel.open();
        ssc.configureBlocking(false); // 设置非阻塞io
        //绑定端口 backlog设为1024
        ssc.socket().bind(new InetSocketAddress(8040),1024000000);
        ssc.register(selector, SelectionKey.OP_ACCEPT);


        while(true){
            System.out.println("阻塞等待...");
            selector.select(); //阻塞等待


            Set<SelectionKey> selectedKeys=selector.selectedKeys();

            Iterator<SelectionKey> iter=selectedKeys.iterator();

            StringBuilder sb=null;
            while (iter.hasNext()){
                SelectionKey selectionKey=iter.next();
                iter.remove(); //移除被选择的通道, 防止重复处理
                if(selectionKey.isAcceptable()){
                    ssc=(ServerSocketChannel) selectionKey.channel();

                    // 需要注册channel
                    SocketChannel sc=ssc.accept();
                    sc.configureBlocking(false);
                    sc.register(selector, SelectionKey.OP_READ);

                    System.out.println("服务端接受连接...");
                }else if(selectionKey.isReadable()){
                    SocketChannel sc=(SocketChannel) selectionKey.channel();

                    int len=-1;

                    ByteArrayOutputStream baos=new ByteArrayOutputStream();

                    while((len=sc.read(buffer))>0){
                        buffer.flip(); // 反转postion, position表示操作的下一个位置
                        
                        byte b=-1;
                        while(buffer.hasRemaining()){
                            b=buffer.get();
                            baos.write(b);
                        }

                        buffer.clear();
                    }
                    if(baos.size()>  0){
                        System.out.println("服务端接收消息为: "+ baos.toString());
                    }
                    if(len==-1){ // -1, 表示客户端主动关闭连接关闭,
                        System.out.println("关闭连接...");
                        sc.close(); // 服务端也需要关闭连接, 否则该链接仍然会可读
                    }
                }
            }
        }

    }

}

public class NioClientTests {
    public static void main(String[] args) throws IOException, InterruptedException {

        SocketChannel sc=SocketChannel.open(new InetSocketAddress(8040));

        sc.configureBlocking(false); // 设置成非阻塞io
        ByteBuffer buffer=ByteBuffer.allocate(102400000);



        int i=0;
        while(true){
            System.out.println("发送消息...");
            buffer.clear(); //清空消息
            buffer.put("timfruit, 您好...".getBytes());

            buffer.flip(); // 将position反转至开始的位置, 用于写
            sc.write(buffer);

            i++;
            if(i>1){
                break;
            }
            Thread.sleep(1000);
        }
        sc.socket().close();
//        sc.close();

    }
}

　　完整源码: 点此查看

学习资料:

　　<深入分析java web技术内幕>

　　UNIX 的5种IO模型介绍　　

   select、poll、epoll之间的区别(搜狗面试)