深入理解java：3. NIO 编程

I/O简介

I/O即输入输出，是计算机与外界世界的一个借口。

IO操作的实际主题是操作系统。

在Java编程中，一般使用流的方式来处理IO，所有的IO都被视作是单个字节的移动，通过stream对象一次移动一个字节。流IO负责把对象转换为字节，然后再转换为对象。

什么是NIO

NIO即New IO，这个库是在JDK1.4中才引入的。NIO主要用到的是块（缓冲），所以NIO的效率要比IO高很多。

在Java API中提供了两套NIO，一套是针对标准输入输出NIO，另一套就是网络编程NIO

NIO和IO最大的区别是数据打包和传输方式。IO是以流的方式处理数据，而NIO是以块（缓冲）的方式处理数据。

面向流的IO一次一个字节的处理数据，一个输入流产生一个字节，一个输出流就消费一个字节。面向流的IO通常处理的很慢。

面向块（缓冲）的IO系统以块的形式处理数据。每一个操作都在一步中产生或消费一个数据块。按块要比按流快的多，但面向块的IO缺少了面向流IO所具有的简单性。

NIO基础

Buffer和Channel是标准NIO中的核心对象（网络NIO中还有个Selector核心对象），几乎每一个IO操作中都会用到它们。

Channel是对原IO中流的模拟，任何来源和目的数据都必须通过一个Channel对象。

一个Buffer实质上是一个容器对象，发给Channel的所有对象都必须先放到Buffer中；同样的，从Channel中读取的任何数据都要读到Buffer中。

使用 Buffer 读写数据一般遵循以下四个步骤：

  1. 写入数据到 Buffer；
  2. 调用 flip() 方法；
  3. 从 Buffer 中读取数据；
  4. 调用 clear() 方法或者 compact() 方法。

clear() 方法会清空整个缓冲区。compact() 方法只会清除已经读过的数据。

Channel是一个对象，可以通过它读取和写入数据。可以把它看做IO中的流：

  1. Channel是双向的，既可以读又可以写，而流是单向的
  2. Channel可以进行异步（非阻塞）的读写
  3. 对Channel的读写必须通过buffer对象

在Java NIO中Channel主要有如下几种类型：

  • FileChannel：从文件读取数据的
  • DatagramChannel：读写UDP网络协议数据
  • SocketChannel：读写TCP网络协议数据
  • ServerSocketChannel：可以监听TCP连接

从文件中读取，需要如下三步：

  1. 从FileInputStream获取Channel
  2. 创建Buffer
  3. 从Channel读取数据到Buffer
public static void copyFileUseNIO(String src,String dst) throws IOException{
           //声明源文件和目标文件
            FileInputStream fi=new FileInputStream(new File(src));
            FileOutputStream fo=new FileOutputStream(new File(dst));
            //获得传输通道channel
            FileChannel inChannel=fi.getChannel();
            FileChannel outChannel=fo.getChannel();
            //获得容器buffer
            ByteBuffer buffer=ByteBuffer.allocate(1024);
            while(true){
                //判断是否读完文件
                int eof =inChannel.read(buffer);
                if(eof==-1){
                    break;  
                }
                //重设一下buffer的position=0，limit=position
                buffer.flip();
                //开始写
                outChannel.write(buffer);
                //写完要重置buffer，重设position=0,limit=capacity
                buffer.clear();
            }
            inChannel.close();
            outChannel.close();
            fi.close();
            fo.close();
}

控制buffer状态的三个变量

  • position：跟踪已经写了多少数据或已经读了多少数据，它指向的是下一个字节来自哪个位置
  • limit：代表还有多少数据可以取出或还有多少空间可以写入，它的值小于等于capacity。
  • capacity：代表缓冲区的最大容量，一般新建一个缓冲区的时候，limit的值和capacity的值默认是相等的。

Selector

前面讲了Java NIO三个核心对象中的Buffer和Channel，现在重点介绍一下第三个核心对象Selector。

Selector是一个对象，它可以注册到很多个Channel上，监听各个Channel上发生的事件，并且能够根据事件情况决定Channel读写。

这样，通过一个线程管理多个Channel，就可以处理大量网络连接了。

下面这幅图展示了一个线程处理3个 Channel的情况：

这里写图片描述

Selector selector = Selector.open();
channel.configureBlocking(false);//异步 非阻塞模式
SelectionKey key =channel.register(selector,SelectionKey.OP_READ);//事件类型

需要注意register()方法的第二个参数，事件类型有四种：

  1. Connect
  2. Accept
  3. Read
  4. Write
某个线程调用select()方法后阻塞了，即使没有通道就绪，也有办法让其从select()方法返回。
只要让其它线程在第一个线程调用select()方法的那个对象上调用Selector.wakeup()方法即可。阻塞在select()方法上的线程会立马返回。

数据处理

Java IO: 从一个阻塞的流中读数据

在IO设计中，我们从InputStream或 Reader逐字节读取数据。假设你正在处理一基于行的文本数据流，例如：

Name: Anna
Age: 25
Email: anna@mailserver.com
Phone: 1234567890

该文本行的流可以这样处理：
InputStream input = … ; // get the InputStream from the client socket

1 BufferedReader reader = new BufferedReader(new InputStreamReader(input));
2  
3 String nameLine   = reader.readLine();
4 String ageLine    = reader.readLine();
5 String emailLine  = reader.readLine();
6 String phoneLine  = reader.readLine();

一旦reader.readLine()方法返回，你就知道肯定文本行就已读完， readline()阻塞直到整行读完，这就是原因。你也知道此行包含名称；

同样，第二个readline()调用返回的时候，你知道这行包含年龄等。

正如你可以看到，该处理程序仅在有新数据读入时运行，并知道每步的数据是什么。一旦正在运行的线程已处理过读入的某些数据，该线程不会再回退数据（大多如此）。

Java NIO:从一个通道里读数据，直到所有的数据都读到缓冲区里.

一个NIO的实现会有所不同，下面是一个简单的例子：

1 ByteBuffer buffer = ByteBuffer.allocate(48);
2  
3 int bytesRead = inChannel.read(buffer);

注意第二行，从通道读取字节到ByteBuffer。

当这个方法调用返回时，你不知道你所需的所有数据是否在缓冲区内。你所知道的是，该缓冲区包含一些字节，这使得处理有点困难。
假设第一次 read(buffer)调用后，读入缓冲区的数据只有半行，例如，“Name:An”，你能处理数据吗？显然不能，需要等待，直到整行数据读入缓存，在此之前，对数据的任何处理毫无意义。

所以，你怎么知道是否该缓冲区包含足够的数据可以处理呢？好了，你不知道。

发现的方法只能查看缓冲区中的数据。其结果是，在你知道所有数据都在缓冲区里之前，你必须检查几次缓冲区的数据。这不仅效率低下，而且可以使程序设计方案杂乱不堪。例如：

1 ByteBuffer buffer = ByteBuffer.allocate(48);
2  
3 int bytesRead = inChannel.read(buffer);
4  
5 while(! bufferFull(bytesRead) ) {
6  
7 bytesRead = inChannel.read(buffer);
8  
9 }

bufferFull()方法必须跟踪有多少数据读入缓冲区，并返回真或假，这取决于缓冲区是否已满。换句话说，如果缓冲区准备好被处理，那么表示缓冲区满了。

bufferFull()方法扫描缓冲区，但必须保持在bufferFull（）方法被调用之前状态相同。如果没有，下一个读入缓冲区的数据可能无法读到正确的位置。这是不可能的，但却是需要注意的又一问题。

如果缓冲区已满，它可以被处理。如果它不满，并且在你的实际案例中有意义，你或许能处理其中的部分数据。但是许多情况下并非如此。

用来处理数据的线程数

NIO可让您只使用一个单线程管理多个通道（网络连接或文件），但付出的代价是解析数据可能会比从一个阻塞流中读取数据更复杂。

如果需要管理同时打开的成千上万个连接，这些连接每次只是发送少量的数据，例如聊天服务器，实现NIO的服务器可能是一个优势。

如果你有少量的连接使用非常高的带宽，一次发送大量的数据，也许典型的IO服务器实现可能非常契合。

基于 java实现TCP/IP+NIO 方式的网络通讯的方法

Java 提供了 SocketChannel和 ServerSocketChannel两个关键的类，

网络 IO 的操作则改为通过ByteBuffer 来实现。

服务器端Java代码 
  1. public static void main(String[] args) throws IOException {  
  2.     System.out.println("Listening for connection on port " + DEFAULT_PORT);  
  3.   
  4.     Selector selector = Selector.open();  
  5.     initServer(selector);  
  6.     //开始监听
  7.     while (true) {  
  8.         selector.select();  //监控所有注册的 channel，当没有可用的IO 操作时会阻塞，有可用的IO 操作时往下执行
  9.   
  10.         for (Iterator<SelectionKey> itor = selector.selectedKeys().iterator(); itor.hasNext();) {  //可进行 IO 操作的 channel的集合：selectedKeys
  11.             SelectionKey key = (SelectionKey) itor.next();  
  12.             itor.remove();  
  13.             try {  
  14.                 if (key.isAcceptable()) {  
  15.                     ServerSocketChannel server = (ServerSocketChannel) key.channel();  //获取Accept操作的Channel
  16.                     SocketChannel client = server.accept();   //客户端的SocketChannel
  17.                     System.out.println("Accepted connection from " + client);  
  18.                     client.configureBlocking(false);  //客户端配置为非阻塞
  19.                     SelectionKey clientKey = client.register(selector, SelectionKey.OP_READ);  //把客户端套接字通道 注册到selector，并注明为OP_READ操作
  20.                     ByteBuffer buffer = ByteBuffer.allocate(100);  
  21.                     clientKey.attach(buffer);  ////客户端SelectionKey附上一个ByteBuffer
  22.                 } else if (key.isReadable()) {  
  23.                     SocketChannel client = (SocketChannel) key.channel();  //获取客户端的SocketChannel
  24.                     ByteBuffer buffer = (ByteBuffer) key.attachment();  buffer .clear();
  25.                     int n = client.read(buffer);  //将客户端套接字通道的数据 读取到缓冲区
  26.                     if (n > 0) {  
  27.                         receiveText = new String(buffer.array(),0,n);//接受到的数据
  28.                         
  29.                         client.register(selector, SelectionKey.OP_WRITE);     // switch to OP_WRITE  
  30.                     }  
  31.                 } else if (key.isWritable()) {  
  32.                     System.out.println("is writable...");  
  33.                     SocketChannel client = (SocketChannel) key.channel();   //获取客户端的SocketChannel
  34.                     ByteBuffer buffer = (ByteBuffer) key.attachment();  buffer.clear();buffer.put(sendText.getBytes()); //sendText ："message from server"
  35.                     buffer.flip(); client.write(buffer);  //输出到通道
  36.                     if (buffer.remaining() == 0) {  // write finished, switch to OP_READ  
  37.                         client.register(selector, SelectionKey.OP_READ);  
  38.                         
  39.                     }  
  40.                 }  
  41.             } catch (IOException e) {  
  42.                 key.cancel();  
  43.                 try { key.channel().close(); } catch (IOException ioe) { }  
  44.             }  
  45.         }  
  46.     }  
  47. }  
  1. private static void initServer(Selector selector) throws IOException,  
  2.             ClosedChannelException {  
  3.         ServerSocketChannel serverChannel = ServerSocketChannel.open();  //打开 服务器套接字 通道
  4.         ServerSocket ss = serverChannel.socket();  //检索 与此服务器套接字通道  关联的套接字
  5.         ss.bind(new InetSocketAddress(DEFAULT_PORT));  //进行服务的绑定
  6.         serverChannel.configureBlocking(false);  //服务器配置为非阻塞
  7.         serverChannel.register(selector, SelectionKey.OP_ACCEPT);  //把服务器套接字通道 注册到selector，并注明为OP_ACCEPT操作
  8.     }

客户端代码：

public class NIOClient {
    /*服务器端地址*/  
    private final static InetSocketAddress SERVER_ADDRESS = new InetSocketAddress(  
            "localhost", 8888);  
  
    public static void main(String[] args) throws IOException {  
        // 打开socket通道  
        SocketChannel clientChannel = SocketChannel.open();  
        // 设置为非阻塞方式  
        clientChannel.configureBlocking(false);  
        // 打开选择器  
        Selector selector = Selector.open();  
        // 注册连接服务端socket动作  
        clientChannel.register(selector, SelectionKey.OP_CONNECT);  
        // 连接  
        clientChannel.connect(SERVER_ADDRESS);  
    
        SocketChannel socketChannel;
        Set<SelectionKey> selectionKeys;    
        while (true) {  
            selector.select();  
            selectionKeys = selector.selectedKeys();    
            for(SelectionKey selectionKey:selectionKeys){ 
                //判断是否为建立连接的事件
                if (selectionKey.isConnectable()) {  
                    System.out.println("client connect");  
                    socketChannel = (SocketChannel) selectionKey.channel(); 
                    // 判断此通道上是否正在进行连接操作。    
                    if (socketChannel.isConnectionPending()) { 
                        //完成连接的建立（TCP三次握手）
                        socketChannel.finishConnect();    
                        sendBuffer.clear();  
                        sendBuffer.put("Hello,Server".getBytes());  
                        sendBuffer.flip();  
                        socketChannel.write(sendBuffer);  
                    }  
                    socketChannel.register(selector, SelectionKey.OP_READ);  
                } else if (selectionKey.isReadable()) {  
                    socketChannel = (SocketChannel) selectionKey.channel();  
                    receiveBuffer.clear();  
                    count=socketChannel.read(receiveBuffer);  
                    if(count>0){  
                        receiveText = new String( receiveBuffer.array(),0,count);
                        socketChannel.register(selector, SelectionKey.OP_WRITE);  
                    }  
                } else if (selectionKey.isWritable()) {  
                    sendBuffer.clear();  
                    socketChannel = (SocketChannel) selectionKey.channel();  
                    sendText = "message from client--" ;  
                    sendBuffer.put(sendText.getBytes());  
                     //将缓冲区各标志复位,因为向里面put了数据标志被改变要想从中读取数据发向服务器,就要复位  
                    sendBuffer.flip();  
                    socketChannel.write(sendBuffer);
                    socketChannel.register(selector, SelectionKey.OP_READ);  
                }  
            }  
            selectionKeys.clear();  
        }  
    }  
}


上面的服务器端代码不够优雅，它将处理服务器Socket和客户连接Socket的代码搅在一起，

这两个Socket通道其实都是注册到Selecto里的，被封装在SelectionKey里面。也就是说无论服务器Socket还是客户连接Socket，都是针对SelectionKey去处理数据的。SelectionKeySelectionKeySelectionKeySelectionKey

当服务器变得复杂，使用命令模式将它们分开变显得非常必要。

首先创建一个接口来抽象对SelectionKey的数据处理：

Java代码 
  1. interface Handler {  
  2.     void execute(Selector selector, SelectionKey key);  
  3. }  

 再来看main函数：

Java代码 
  1. public static void main(String[] args) throws IOException {  
  2.     System.out.println("Listening for connection on port " + DEFAULT_PORT);  
  3.   
  4.     Selector selector = Selector.open();  
  5.     initServer(selector);  
  6.   
  7.     while (true) {  
  8.         selector.select();  
  9.   
  10.         for (Iterator<SelectionKey> itor = selector.selectedKeys().iterator(); itor.hasNext();) {  
  11.             SelectionKey key = (SelectionKey) itor.next();  
  12.             itor.remove();  
  13.             Handler handler = (Handler) key.attachment();  //利用Java的多态特性，动态处理serverSocket SelectionKey 或 ClientSocket SelectionKey
  14.             handler.execute(selector, key);  
  15.         }  
  16.     }  
  17. }  
  18.   
  19. private static void initServer(Selector selector) throws IOException,  
  20.         ClosedChannelException {  
  21.     ServerSocketChannel serverChannel = ServerSocketChannel.open();  
  22.     ServerSocket ss = serverChannel.socket();  
  23.     ss.bind(new InetSocketAddress(DEFAULT_PORT));  
  24.     serverChannel.configureBlocking(false);  
  25.     SelectionKey serverKey = serverChannel.register(selector, SelectionKey.OP_ACCEPT);  //serverChannel注册到selector后，生成一个serverSocket SelectionKey
  26.     serverKey.attach(new ServerHandler());  //在serverSocket SelectionKey上附加了一个ServerHandler的对象，进行serverSocket SelectionKey的数据处理去处理
  27. }  

下面是ServerHandler的代码： 在获取serverSocket SelectionKey时需要进行的数据处理

Java代码 
  1. class ServerHandler implements Handler {  
  2.     public void execute(Selector selector, SelectionKey key) {  
  3.         ServerSocketChannel server = (ServerSocketChannel) key.channel();  
  4.         SocketChannel client = null;  
  5.         try {  
  6.             client = server.accept();  
  7.             System.out.println("Accepted connection from " + client);  
  8.         } catch (IOException e) {  
  9.             e.printStackTrace();  
  10.             return;  
  11.         }  
  12.           
  13.         SelectionKey clientKey = null;  
  14.         try {  
  15.             client.configureBlocking(false);  
  16.             clientKey = client.register(selector, SelectionKey.OP_READ);  
  17.             clientKey.attach(new ClientHandler());  //在clientSocket SelectionKey上附加了一个clientHandler的对象，进行clientSocket SelectionKey的数据处理
  18.         } catch (IOException e) {  
  19.             if (clientKey != null)  
  20.                 clientKey.cancel();  
  21.             try { client.close(); } catch (IOException ioe) { }  
  22.         }  
  23.     }  
  24. }  

下面是ClientHandler的代码： 在获取ClientSocket SelectionKey时需要进行的数据处理

Java代码  收藏代码
  1. class ClientHandler implements Handler {  
  2.     private ByteBuffer buffer;  
  3.       
  4.     public ClientHandler() {  
  5.         buffer = ByteBuffer.allocate(100);  
  6.     }  
  7.       
  8.     public void execute(Selector selector, SelectionKey key) {  
  9.         try {  
  10.             if (key.isReadable()) {  
  11.                 readKey(selector, key);  
  12.             } else if (key.isWritable()) {  
  13.                 writeKey(selector, key);  
  14.             }  
  15.         } catch (IOException e) {  
  16.             key.cancel();  
  17.             try { key.channel().close(); } catch (IOException ioe) { }  
  18.         }  
  19.     }  
  20.       
  21.     private void readKey(Selector selector, SelectionKey key) throws IOException {  
  22.         SocketChannel client = (SocketChannel) key.channel();  
  23.         int n = client.read(buffer);  
  24.         if (n > 0) {  
  25.             buffer.flip();  
  26.             key.interestOps(SelectionKey.OP_WRITE);     // switch to OP_WRITE  
  27.         }  
  28.     }  
  29.       
  30.     private void writeKey(Selector selector, SelectionKey key) throws IOException {  
  31.         // System.out.println("is writable...");  
  32.         SocketChannel client = (SocketChannel) key.channel();  
  33.         client.write(buffer);  
  34.         if (buffer.remaining() == 0) {  // write finished, switch to OP_READ  
  35.             buffer.clear();  
  36.             key.interestOps(SelectionKey.OP_READ);  
  37.         }  
  38.     }  
  39. }  

代码结构显得更清晰。

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/my376908915/p/6767922.html