java nio 学习

Posted 2020-08-27 liouwei4083

     1,nio概述

     NIO主要有三大核心部分：Channel(通道)，Buffer(缓冲区), Selector。传统IO基于字节流和字符流进行操作，而NIO基于Channel和Buffer(缓冲区)进行操作，数据总是从通道读取到缓冲区中，或者从缓冲区写入到通道中。Selector(选择区)用于监听多个通道的事件（比如：连接打开，数据到达）。因此，单个线程可以监听多个数据通道。

 

NIO和传统IO（一下简称IO）之间第一个最大的区别是，IO是面向流的，NIO是面向缓冲区的。 Java IO面向流意味着每次从流中读一个或多个字节，直至读取所有字节，它们没有被缓存在任何地方。此外，它不能前后移动流中的数据。如果需要前后移动从流中读取的数据，需要先将它缓存到一个缓冲区。NIO的缓冲导向方法略有不同。数据读取到一个它稍后处理的缓冲区，需要时可在缓冲区中前后移动。这就增加了处理过程中的灵活性。但是，还需要检查是否该缓冲区中包含所有您需要处理的数据。而且，需确保当更多的数据读入缓冲区时，不要覆盖缓冲区里尚未处理的数据。

 

IO的各种流是阻塞的。这意味着，当一个线程调用read() 或 write()时，该线程被阻塞，直到有一些数据被读取，或数据完全写入。该线程在此期间不能再干任何事情了。 NIO的非阻塞模式，使一个线程从某通道发送请求读取数据，但是它仅能得到目前可用的数据，如果目前没有数据可用时，就什么都不会获取。而不是保持线程阻塞，所以直至数据变的可以读取之前，该线程可以继续做其他的事情。 非阻塞写也是如此。一个线程请求写入一些数据到某通道，但不需要等待它完全写入，这个线程同时可以去做别的事情。 线程通常将非阻塞IO的空闲时间用于在其它通道上执行IO操作，所以一个单独的线程现在可以管理多个输入和输出通道（channel）。

 

     2,nio相关类

     buffer

     nio中的数据读写操作都要在buffer中进行,buffer是一个线性的有序的数据集,只能容纳某种特定的数据类型.

     buffer中有三个变量在读取的时候会涉及到,position,limit,capacity

 

 

示例代码如下

        IntBuffer buf = IntBuffer.allocate(10) ;    // 准备出10个大小的缓冲区
        System.out.print("1、写入数据之前的position、limit和capacity：") ;
        System.out.println("position = " + buf.position() + "，limit = " + buf.limit() + "，capacty = " + buf.capacity()) ;
        int temp[] = {5,7,9} ;// 定义一个int数组
        buf.put(3) ;    // 设置一个数据
        buf.put(temp) ;    // 此时已经存放了四个记录
        System.out.print("2、写入数据之后的position、limit和capacity：") ;
        System.out.println("position = " + buf.position() + "，limit = " + buf.limit() + "，capacty = " + buf.capacity()) ;
 
        buf.flip() ;    // 重设缓冲区
        // postion = 0 ,limit = 原本position
        System.out.print("3、准备输出数据时的position、limit和capacity：") ;
        System.out.println("position = " + buf.position() + "，limit = " + buf.limit() + "，capacty = " + buf.capacity()) ;
        System.out.print("缓冲区中的内容：") ;
        while(buf.hasRemaining()){
            int x = buf.get() ;
            System.out.print(x + "、") ;
        }

 

buffer示意图如下

 

 

 

     channel

     channel可以用来读取和写入数据,类似于之前的inputstream和outputstream,但是操作的数据都是先读写到buffer中的,或者从buffer中来读写.,传统的流操作分为input和output,而channel既可以输入也可以输出.下图展示了channel的类体系图,其中常用的红框标注

     

 

 

     filechannel的例子

     

        File file1 = new File("d:" + File.separator + "note.txt") ;
        File file2 = new File("d:" + File.separator + "outnote.txt") ;
        FileInputStream input = null ;
        FileOutputStream output = null ;
        output = new FileOutputStream(file2) ;
        input = new FileInputStream(file1) ;
        FileChannel fout = null ;    // 声明FileChannel对象
        FileChannel fin = null ;    // 定义输入的通道
        fout = output.getChannel() ;    // 得到输出的通道
        fin = input.getChannel() ;    // 得到输入的通道
        ByteBuffer buf = ByteBuffer.allocate(1024) ;
 
        int temp = 0 ;
        while((temp=fin.read(buf))!=-1){
            buf.flip() ;
            fout.write(buf) ;
            buf.clear() ;    // 清空缓冲区,所有的状态变量的位置恢复到原点
        }

 

 

     selector

     在nio中selector是最重要的一个概念,在原来使用io和scoket的构造网络服务的时候,所有的网络服务将使用阻塞的方法来进行客户端的链接,如果使用了selector就可以构造一个非阻塞的网络服务.

     在构造非阻塞网络服务的时候,需要使用selectablechannel类想selector类注册,而且在新io中实现网络服务需要依靠serverscoketchannel和scoketchannel这两个类都是selectablechannel的子类,这个父类提供了注册selector和阻塞模式的选择.

     

 

    

     

     3,scoket网络应用中的nio实现

     下面通过代码来实现三种现在网络中使用的网络服务提供模式(时间服务器)

          传统的bio

          传统的bio模型,服务端一般都是一个独立的acceptor线程负责监听客户端的连接,收到客户端的请求后,为每一个client都开启一个线程去进行处理处理完成后通过outputstream返回给client,线程销毁.如下图

 

 

缺点就是缺乏弹性伸缩能力,当client端的并发量上去之后,server这边的线程个数也上去了,慢慢就会导致系统性能下降,堆栈溢出,创建新线程失败,系统宕机等.

实例代码

     

public class TimeServer {
    public static void main(String[] args) throws IOException {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
            }
        }
        ServerSocket server = null;
        try {
            server = new ServerSocket(port);
            System.out.println("The time server is start in port : " + port);
            Socket socket = null;
            while (true) {
                socket = server.accept();
                new Thread(new TimeServerHandler(socket)).start();
            }
        } finally {
            if (server != null) {
                System.out.println("The time server close");
                server.close();
                server = null;
            }
        }
    }
}
 
public class TimeServerHandler implements Runnable {
    private Socket socket;
    public TimeServerHandler(Socket socket) {
        this.socket = socket;
    }
    public void run() {
        BufferedReader in = null;
        PrintWriter out = null;
        try {
            in = new BufferedReader(new InputStreamReader(
                    this.socket.getInputStream()));
            out = new PrintWriter(this.socket.getOutputStream(), true);
            String currentTime = null;
            String body = null;
            while (true) {
                body = in.readLine();
                if (body == null)
                    break;
                System.out.println("The time server receive order : " + body);
                currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(
                        System.currentTimeMillis()).toString() : "BAD ORDER";
                out.println(currentTime);
            }
 
        } catch (Exception e) {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
            if (out != null) {
                out.close();
                out = null;
            }
            if (this.socket != null) {
                try {
                    this.socket.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
                this.socket = null;
            }
        }
    }
}

 

 

 

          pio

          伪异步io模式,伪了解决上面的那种问题,对她进行优化,在server端,使用线程池的来处理client的链接,这样就可以防止因为创建了太多的线程吧系统给拖垮的问题,虽然解决了这个,但是底层采用的依然是阻塞的网络服务模型,当client数量上来的时候,或者一个线程的处理时间过长,线程池一直处于满载的状态,后续的client会一直进不来,连接超时. 下面是实例

     

public class TimeServer {
    public static void main(String[] args) throws IOException {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
            }
        }
        ServerSocket server = null;
        try {
            server = new ServerSocket(port);
            System.out.println("The time server is start in port : " + port);
            Socket socket = null;
            TimeServerHandlerExecutePool singleExecutor = new TimeServerHandlerExecutePool(
                    50, 10000);// 创建IO任务线程池
            while (true) {
                socket = server.accept();
                singleExecutor.execute(new TimeServerHandler(socket));
            }
        } finally {
            if (server != null) {
                System.out.println("The time server close");
                server.close();
                server = null;
            }
        }
    }
}
 
 
public class TimeServerHandlerExecutePool {
 
    private ExecutorService executor;
 
    public TimeServerHandlerExecutePool(int maxPoolSize, int queueSize) {
    executor = new ThreadPoolExecutor(Runtime.getRuntime()
        .availableProcessors(), maxPoolSize, 120L, TimeUnit.SECONDS,
        new ArrayBlockingQueue<java.lang.Runnable>(queueSize));
    }
 
    public void execute(java.lang.Runnable task) {
    executor.execute(task);
    }
}

 

 

          nio

     通俗来将,其实就是,我服务端有个监听8080端口的serverscoket,我吧这个它和一个serverscoketchannel关联到一起,以后给她的信息也好,他要输出的信息也好,都去操作这个channel,然后,我把这个channel注册到一个selector上,并且选择一个事件,(on_accpet),这样类似的也可以在这个selector上注册其他的一些channel并选择好对应的事件类型,然后,这个selector它自己会去无限轮询所有的channel,然后看看每一个channel是否已经达到了那个事件的状态,比如,我的那个serverscoket,如果有一个client链接过来了,那么这个channel就已经是on_accpet状态了,当selector轮询到这个channel的时候,就要开始处理我这个channel了,针对这种serverscoket的处理就是,先通过selector拿到当状态的selectkey,从selectkey中拿到当状态的channel,然后从这个serverscoketchannel中拿到client端对应的scoketchannel(普通的io是话是拿到一个scoket),拿到这个scoketchannel后,立马把这个channel也注册到selector里面,设置对应事件类型为on_read意思就是,client如果传输数据过来的状态,然后selector记录做轮询,又一个client链接过来,就又有一个scoketchannel出来,病注册到selector里,轮询的时候,如果发现有on_read事件了,那么就可以拿到这个scoketchannel,然后read里面的信息,并做处理,并write对应是信息,如果处理完了,就把对应的该关的都关掉.   这样一来不论client有多少个,server这边都可以处理的来,因为server这边其实就是一个selector线程在做轮询处理而已,  如果还需要优化,那就是在这个的基础上,吧selector也做成线程池模型.  整体理解下来就是这样.   这样就肯定不会有client链接不过来的情况. 下面看书本上咋说

 

     

 

 

 

 

 

下面是实例代码 可以研究研究

public class TimeServer {
 
    public static void main(String[] args) throws IOException {
    int port = 8080;
    if (args != null && args.length > 0) {
        try {
        port = Integer.valueOf(args[0]);
        } catch (NumberFormatException e) {
        // 采用默认值
        }
    }
    MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
    new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();
    }
}
 
 
public class MultiplexerTimeServer implements Runnable {
    private Selector selector;
    private ServerSocketChannel servChannel;
    private volatile boolean stop;
    public MultiplexerTimeServer(int port) {
        try {
            selector = Selector.open();
            servChannel = ServerSocketChannel.open();
            servChannel.configureBlocking(false);
            servChannel.socket().bind(new InetSocketAddress(port), 1024);
            servChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("The time server is start in port : " + port);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
    public void run() {
        while (!stop) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.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();
            }
        }
        // 多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源
        if (selector != null)
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
    }
    private void handleInput(SelectionKey key) throws IOException {
        if (key.isValid()) {
            // 处理新接入的请求消息
            if (key.isAcceptable()) {
                // Accept the new connection
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                // Add the new connection to the selector
                sc.register(selector, SelectionKey.OP_READ);
            }
            if (key.isReadable()) {
                // Read the data
                SocketChannel sc = (SocketChannel) key.channel();
                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("The time server receive order : "
                            + body);
                    String currentTime = "QUERY TIME ORDER"
                            .equalsIgnoreCase(body) ? new java.util.Date(
                            System.currentTimeMillis()).toString()
                            : "BAD ORDER";
                    doWrite(sc, currentTime);
                } else if (readBytes < 0) {
                    // 对端链路关闭
                    key.cancel();
                    sc.close();
                } else
                    ; // 读到0字节，忽略
            }
        }
    }
    private void doWrite(SocketChannel channel, String response)
            throws IOException {
        if (response != null && response.trim().length() > 0) {
            byte[] bytes = response.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            channel.write(writeBuffer);
        }
    }
    public void stop() {
        this.stop = true;
    }
}
 

 

 

          aio     

     java nio2.0引入了异步通道的概念提供了异步文件通道和异步套接字通道的实现,上面说的那个nio其实是同步非阻塞io,aio才是真正的异步非阻塞io,还没有研究........