java基础知识--NIO详解及实战

Posted 2021-05-15 JordanInShenzhen

前面已经介绍了java常见的几种IO操作，包括文件读写、缓冲区的操作等等。这些操作虽然都是我们日常在写业务代码的时候常用的操作，但是还有一种在互联网中也是极其重要的IO，那就是NIO。NIO是jdk1.4版本推出来的新功能，主要是为了解决在网络IO的过程中，同步IO阻塞导致系统资源浪费的场景。所以很多人说NIO，其实都是特指AIO，即异步IO

在介绍NIO（AIO）之前，我们先来看看原来的BIO有什么问题。现在我们来模拟这么一个互联网提供服务的场景，一个服务端，通过监听某个端口给客户端提供服务，为了能够达到一个服务端能够服务多客户端的能力，当有客户端请求连接过来的时候，通过开启一个线程方式去处理这个请求。服务端代码如下：

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ServerSocket;
import java.net.Socket;

public class Test {
	public static void main(String[] args) throws IOException {
		int port = 9010;
		try(ServerSocket s= new ServerSocket(port)){
			while(true) {
				Socket socket = s.accept();
				// 开一个线程去处理当前请求
				new Thread(new SocketProcess(socket)).start();
			}
		}catch(Exception e) {
			// 异常处理
			e.printStackTrace();
		}
	}
	
	static class SocketProcess implements Runnable{
		Socket socket;
		public SocketProcess(Socket socket) {
			super();
			this.socket = socket;
		}
		
		@Override
		public void run() {
			try(BufferedReader reader = new BufferedReader(
					new InputStreamReader(socket.getInputStream()))) {
				// 接收打印数据
				String message = null;
				while ((message = reader.readLine()) != null) {
					System.out.println(message);
				}
			} catch (Exception e) {
				// 异常处理
				e.printStackTrace();
			}
		}
	}
}

上面这段代码，我们为了达到一个服务器，处理多个请求的效果，通过每次来一个请求，都new一个线程去处理，也能达到提升服务器资源利用率的效果。但是，这里还是存在一个问题无法解决，那就是在下面这行代码：

BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()))

当我们已经开启一个线程去处理用户的输入请求的时候，我们会发现，我们需要等待用户的输入，如果用户一直不输入，这个线程会一直阻塞在这里，从而导致线程处于空闲状态。这种空闲状态越多的时候，我们的资源就会越被这些空闲的线程所占用（每个线程都有他自己的线程栈，最大1M），即使使用了线程池，也无法解决线程被等待用户输入挂起所导致的问题（把请求放入等待队列会导致真正想要处理输入的客户端响应非常慢，这在现实生活中是不可接受的）。

为了解决上述问题，NIO（AIO）问世了

NIO基础

在介绍NIO之前，先介绍NIO的三个核心对象:Buffer、channel、selector。

Buffer

Buffer是一个对象，它包含一些要写入或读出的数据。在NIO中，数据是放入buffer对象的，而在IO中，数据是直接写入或者读到Stream对象的。应用程序不能直接对 Channel 进行读写操作，而必须通过 Buffer 来进行，即 Channel 是通过 Buffer 来读写数据的。在NIO中，所有的数据都是用Buffer处理的，它是NIO读写数据的中转池。

Buffer底层是一个数组，通常是一个字节数组，但也可以是其他类型的数组。但一个缓冲区不仅仅是一个数组，重要的是它提供了对数据的结构化访问，而且还可以跟踪系统的读写进程。

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

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

可能很多人看上面这些步骤看的有点懵，那么我们先来看看这些方法以及Buffer的底层实现原理，了解完我们再回头来看这些方法调用顺序就会一目了然

Buffer底层是一个数组，可以是ByteBuffer、CharBuffer、IntBuffer、FloatBuffer甚至DoubleBuffer等等。但是它并没有把这个数组直接暴露给用户直接读写，而是封装了一系列的方法，用户通过调用这些方法来操作数组的读写位置，这些方法主要控制以下三个变量来控制的：

1. position：跟踪已经写了多少数据或读了多少数据，它指向的是下一个字节来自哪个位置

2. limit：代表最多可以取出或写入多少数据，它的值小于等于capacity。

3. capacity：代表缓冲区的最大容量，一般新建一个缓冲区的时候，limit的值和capacity的值默认是相等的。

4. mark：标记，调用mark()会设置mark=position，再调用reset()可以让position恢复到标记的位置

介绍完上面四个控制buffer读写位置的变量后，我们来看下flip()方法底层实现原理：

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

通过上面flip的源码我们可以发现，其实它主要是将当前读到的字节所在的位置，设置成了limit，然后将下一次开始读写的位置设置成从0开始。这意味着什么，是不是意味着下一次的操作，我们将把上一次写进来（假如上次是写数据到buffer）的数据，全都一个一个输出（读出）到目标去？所以，每次我们再写完数据到buffer之后，都会调用flip方法，将当前的写模式，切换到读模式。

接下来，我们再来看下clear()方法

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

通过上面我们看clear方法的源码，我们可以知道，clear方法是讲当前位置置成0，并把limit放到buffer最大位置，这意味着如果我们把缓冲区的数据全都读出到目标后，这个方法就重设了缓冲区以便接收更多的字节。实际上Buffer中的数据并未清除，只是这些标记告诉我们可以从哪里开始往Buffer里写数据。

最后，我们来看下compact方法

官方文档上介绍该方法为压缩缓冲区，主要动作是把从position到limit中的内容移到0到limit-position的区域内，position和limit的取值也分别变成limit-position、capacity。如果先将positon读写到limit，再compact，那么相当于clear()。下面这两张图清晰展示了compact方法使用前后的效果：

假如compact方法调用前如下：

则调用compact方法后，缓冲区的情况：

其实简而言之，就是清空已经读过的数据，把position设置到最后一个未读数据的后面，并将limit置成capacity。这样，就将当前buffer的读模式切换到写模式，把当前要写入的数据，写到上次limit的后面，这样读入的数据就不会造成数据的覆盖或者丢失。

好，现在我们再去看上面那个过程，就一目了然了，从将待读取的数据写入到缓冲区，然后切换读写开始位置，并将limit设置到第一步写入截止位置，然后开始读取到目的地；读完后，清空缓冲区，给下次读写使用。

buffer常用的方法
limit(), limit(10)：其中读取和设置这4个属性的方法的命名和jQuery中的val(),val(10)类似，一个负责get，一个负责set
reset()：把position设置成mark的值，相当于之前做过一个标记，现在要退回到之前标记的地方
rewind()：把position设为0，mark设为-1，不改变limit的值
remaining()：return limit - position;返回limit和position之间相对位置差
hasRemaining() ：return position < limit返回是否还有未读内容
get()：相对读，从position位置读取一个byte，并将position+1，为下次读写作准备
get(int index)：绝对读，读取byteBuffer底层的bytes中下标为index的byte，不改变position
get(byte[] dst, int offset, int length)：从position位置开始相对读，读length个byte，并写入dst下标从offset到offset+length的区域
put(byte b)：相对写，向position的位置写入一个byte，并将postion+1，为下次读写作准备
put(int index, byte b)：绝对写，向byteBuffer底层的bytes中下标为index的位置插入byte b，不改变position
put(ByteBuffer src)：用相对写，把src中可读的部分（也就是position到limit）写入此byteBuffer
put(byte[] src, int offset, int length)：从src数组中的offset到offset+length区域读取数据并使用相对写写入此byteBuffer
 

channel

Channel是一个对象，可以通过它读取和写入数据。channel和java IO中的流比较像，但是它还有如下特点：

1. Channel是双向的，既可以读又可以写，而流是单向的

2. Channel可以进行异步的读写

v3. 对Channel的读写必须通过buffer对象

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

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

Selector

selector是异步IO的核心，它是一个对象，它可以注册到很多个Channel上，监听各个Channel上发生的事件，并且能够根据自己感兴趣的事件情况决定Channel读写。这样，通过一个线程管理多个Channel，就可以处理大量网络连接了。这一点，是NIO处理互联网上大并发请求的一个利器

利用selector，我们可以通过利用一个线程来处理所有的channels。线程之间的切换对操作系统来说代价是很高的，并且每个线程也会占用一定的系统资源。所以，对系统来说使用的线程越少越好。一般来说，线程的个数设定成跟CPU核数保持一致，这样能够省去线程之间的切换。

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

创建一个selector并将Channel注册到selector上：

Selector selector = Selector.open();
channel.configureBlocking(false);
SelectionKey key =channel.register(selector,SelectionKey.OP_READ);

上面代码中，我们除了创建selector、将channel注册到selector上外，还把channel设置成非阻塞模式，否则异步IO就无法工作，这就意味着我们不能把一个FileChannel注册到Selector，因为FileChannel没有异步模式，但是网络编程中的SocketChannel是可以的。

注意第三行代码，channel注册到selector上的时候，需要指定selector对channel上感兴趣的事件的种类。SelectionKey中定义的事件常量有如下四种：

SelectionKey.OP_CONNECT

SelectionKey.OP_ACCEPT

SelectionKey.OP_READ

SelectionKey.OP_WRITE

通道触发了一个事件意思是该事件已经 Ready(就绪)。所以，某个Channel成功连接到另一个服务器称触发事件是OP_CONNECT。一个ServerSocketChannel准备好接收新连接触发的事件是OP_ACCEPT，一个有数据可读的通道触发的事件是OP_READ，等待写数据的通道触发的事件是OP_WRITE。

如果你对多个事件感兴趣，可以通过"|"操作符来连接这些常量：

int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE; 

SelectionKey 代表这个通道在此 Selector 上的这个注册。当某个 Selector 通知你某个传入事件时，它是通过提供对应于该事件的 SelectionKey 来进行的。SelectionKey 还可以用于取消通道的注册。SelectionKey中包含如下属性：interest set、ready set、Channel、Selector、attach

Interest Set

就像我们在前面讲到的把Channel注册到Selector来监听感兴趣的事件，interest set就是你要选择的感兴趣的事件的集合。你可以通过SelectionKey对象来读写interest set，我们可以通过用AND 和SelectionKey 中的常量做运算，从SelectionKey中找到我们感兴趣的事件。:

int interestSet = selectionKey.interestOps();
boolean isInterestedInAccept  = interestSet & SelectionKey.OP_ACCEPT;
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;
boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;   

Ready Set

ready set 是通道已经准备就绪的操作的集合。在一次选Selection之后，你应该会首先访问这个ready set。可以用像检测interest集合那样的方法，来检测Channel中什么事件或操作已经就绪。

int readySet = selectionKey.readyOps();

当然也可以使用以下四个方法，它们都会返回一个布尔类型（这种用法挺多的，简洁）：

selectionKey.isAcceptable();
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();

Channel 和Selector

我们可以通过SelectionKey获得Selector和注册的Channel：

Channel  channel  = selectionKey.channel();
Selector selector = selectionKey.selector(); 

通过Selector选择通道

一旦向Selector注册了一或多个通道，就可以调用几个重载的select()方法。这些方法返回你所感兴趣的事件（如连接、接受、读或写）已经准备就绪的那些通道。换句话说，如果你对“Read Ready”的通道感兴趣，select()方法会返回读事件已经就绪的那些通道：

int select()： 阻塞到至少有一个通道在你注册的事件上就绪
int select(long timeout)：select()一样，只不过最长只会阻塞timeout毫秒(参数)
int selectNow()： 不会阻塞，不管什么通道就绪都立刻返回，此方法执行非阻塞的选择操作。如果自从前一次选择操作后，没有通道变成可选择的，则此方法直接返回零。

select()方法返回的int值表示有多少通道已经就绪。亦即，自上次调用select()方法后有多少通道变成就绪状态。如果调用select()方法，因为有一个通道变成就绪状态，返回了1，若再次调用select()方法，如果另一个通道就绪了，它会再次返回1。如果对第一个就绪的channel没有做任何操作，现在就有两个就绪的通道，但在每次select()方法调用之间，只有一个通道处于就绪状态。
selectedKeys()

划重点，我们把一堆的channel注册到selector上去后，如果我们通过调用了select()方法，它就会返回一个数值，表示一个或多个通道已经就绪，然后你就可以通过调用selector.selectedKeys()方法得到SelectionKey集合，从而获得就绪的Channel：

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

好，接下来，我们可以写一个基于NIO的服务端了代码了（时间有点晚了，代码比较粗，下次整一个多线程处理单接口的）
 

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
 
public class NiosServer implements Runnable{
    private int port = 8888;
    private Charset cs = Charset.forName("gbk"); 
    private ByteBuffer sBuffer = ByteBuffer.allocate(1024);  
    private ByteBuffer rBuffer = ByteBuffer.allocate(1024);  
    private Map<String, SocketChannel> clientsMap = new HashMap<String, SocketChannel>();  
    private Selector selector;  
       
    public NIOSServer(int port){  
        this.port = port;  
        try {  
            init();  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
	}
    
    private void init() throws IOException{  
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();  
        serverSocketChannel.configureBlocking(false);  
        ServerSocket serverSocket = serverSocketChannel.socket();  
        serverSocket.bind(new InetSocketAddress(port));  
        selector = Selector.open();
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);  
        System.out.println("server start on port:"+port);  
    }  
 
    private void listen(){   
    } 
     
    @Override
    public void run() {
    	while (true) {  
            try {  
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();  
                for(SelectionKey key : selectionKeys){  
                    handle(key);  
                }  
                selectionKeys.clear();
            } catch (Exception e) {  
                e.printStackTrace();  
                break;  
            }  
               
        }  
    } 
    private void handle(SelectionKey selectionKey) throws IOException, InterruptedException {  
        ServerSocketChannel server = null;  
        SocketChannel client = null;  
        String receiveText=null;  
        int count=0;  
        if (selectionKey.isAcceptable()) {  
            server = (ServerSocketChannel) selectionKey.channel();  
            client = server.accept();
            //判断client 是否为空
            if(client != null){
                client.configureBlocking(false);  
                client.register(selector, SelectionKey.OP_READ);
            }
            // Thread.sleep(10*1000);
        } else if (selectionKey.isReadable()) {  
            client = (SocketChannel) selectionKey.channel();  
            rBuffer.clear();  
            count = client.read(rBuffer);  
            if (count > 0) {  
                rBuffer.flip();  
                receiveText = String.valueOf(cs.decode(rBuffer).array());  
                System.out.println(client.toString()+":"+receiveText);  
                dispatch(client, receiveText);  
                client = (SocketChannel) selectionKey.channel();  
                client.register(selector, SelectionKey.OP_READ);  
            }  
        }   
    }  
       
    private void dispatch(SocketChannel client,String info) throws IOException{  
        Socket s = client.socket();  
        String name = "["+s.getInetAddress().toString().substring(1)+":"+Integer.toHexString(client.hashCode())+"]";  
        if(!clientsMap.isEmpty()){  
            for(Map.Entry<String, SocketChannel> entry : clientsMap.entrySet()){  
                SocketChannel temp = entry.getValue();  
                if(!client.equals(temp)){  
                    sBuffer.clear();  
                    sBuffer.put((name+":"+info).getBytes());  
                    sBuffer.flip();    
                    temp.write(sBuffer);  
                }  
            }  
        }  
        clientsMap.put(name, client);  
    }  
     
    public static void main(String[] args) throws InterruptedException {
        //创建NIOSServer
        NIOSServer server1 = new NIOSServer(7778);  
        NIOSServer server2 = new NIOSServer(7777);  
        new Thread(server1).start();
        new Thread(server2).start();
    }
}