NIO之阻塞IO与非阻塞IO（包含Selector使用）

阻塞IO

　　传统的 IO 流都是阻塞式的。

　　也就是说，当一个线程调用 read() 或 write()时，该线程被阻塞，直到有一些数据被读取或写入，该线程在此期间不能执行其他任务。

　　因此，在完成网络通信进行 IO 操作时，由于线程会阻塞，所以服务器端必须为每个客户端都提供一个独立的线程进行处理，当服务器端需要处理大量客户端时，性能急剧下降。

　　注意：在阻塞IO操作的过程中，用来提高程序的解决方案一般是使用多线程来处理，但是开辟线程也是比较耗费资源的。

阻塞代码-远程文件传输

服务端无返回代码

public static void main(String[] args){
        new Thread(new Runnable() {
            @Override
            public void run() {
                server();
            }
        }).start();
        
        new Thread(new Runnable() {
            @Override
            public void run() {
                client();
            }
        }).start();
    }
    
    // 阻塞io,客户端
    public static void client(){
        
        SocketChannel socketChannel = null;// 网络channel
        FileChannel inChannel = null;// 本地channel
        try {
            // 1.获取通道
            socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 2653));
            inChannel = FileChannel.open(Paths.get("d:\apiclient.txt"), StandardOpenOption.READ);
            
            // 2.分配缓冲区
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            
            // 3.本地通道数据通过缓冲区传输至网络channel
            while (inChannel.read(byteBuffer)!=-1) {                
                byteBuffer.flip();// 切换写模式
                socketChannel.write(byteBuffer);
                byteBuffer.clear();
            }
            System.out.println("client local channel to socket channel");
            
        } catch (IOException ex) {
            ex.printStackTrace();
        } finally{
            if (socketChannel!=null) {
                try {
                    socketChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
            if (inChannel!=null) {
                try {
                    inChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
    // 阻塞io,服务端
    public static void server(){
        ServerSocketChannel serverSocketChannel = null;// 网络channel
        SocketChannel socketChannel = null;// 接收客户端的channel
        FileChannel outChannel = null;// 本地channel
        try {
            // 1.获取通道
            serverSocketChannel = ServerSocketChannel.open();// serverSocket.getChannel();
            outChannel = FileChannel.open(Paths.get("d:\apiserver.txt"), StandardOpenOption.WRITE, StandardOpenOption.CREATE);
            // 2.连接
            serverSocketChannel.bind(new InetSocketAddress(2653));
            // 3.获取客户端连接通道
            socketChannel = serverSocketChannel.accept();
            System.out.println("accept client success");
            // 4.分配缓冲区
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            // 5.接收客户端数据保存至本地
            while (socketChannel.read(byteBuffer)!=-1) {                
                byteBuffer.flip();
                outChannel.write(byteBuffer);
                byteBuffer.clear();
            }
            System.out.println("socket channel to server local channel");
            
        } catch (IOException ex) {
            ex.printStackTrace();
        }finally{
            if (serverSocketChannel!=null) {
                try {
                    serverSocketChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
            if (socketChannel!=null) {
                try {
                    socketChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
            if (outChannel!=null) {
                try {
                    outChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }

 服务端有返回代码

public static void main(String[] args){
        new Thread(new Runnable() {
            @Override
            public void run() {
                server();
            }
        }).start();
        
        new Thread(new Runnable() {
            @Override
            public void run() {
                client();
            }
        }).start();
    }
    
    // 阻塞io,客户端
    public static void client(){
        
        SocketChannel socketChannel = null;// 网络channel
        FileChannel inChannel = null;// 本地channel
        try {
            // 1.获取通道
            socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 2653));
            inChannel = FileChannel.open(Paths.get("d:\apiclient.txt"), StandardOpenOption.READ);
            
            // 2.分配缓冲区
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            
            // 3.本地通道数据通过缓冲区传输至网络channel
            while (inChannel.read(byteBuffer)!=-1) {                
                byteBuffer.flip();// 切换写模式
                socketChannel.write(byteBuffer);
                byteBuffer.clear();
            }
            System.out.println("client local channel to socket channel");
            
            // 4.切换到接收数据的模式
            socketChannel.shutdownOutput();
            System.out.println("client wait for server's message...");
            
            // 5.接收服务端的数据
            int len;
            while ((len=socketChannel.read(byteBuffer))!=-1) {               
                byteBuffer.flip();
                System.out.println("client receive server msg:"+new String(byteBuffer.array(), 0, len));
                byteBuffer.clear();
            }
            
        } catch (IOException ex) {
            ex.printStackTrace();
        } finally{
            if (socketChannel!=null) {
                try {
                    socketChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
            if (inChannel!=null) {
                try {
                    inChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
    // 阻塞io,服务端
    public static void server(){
        ServerSocketChannel serverSocketChannel = null;// 网络channel
        SocketChannel socketChannel = null;// 接收客户端的channel
        FileChannel outChannel = null;// 本地channel
        try {
            // 1.获取通道
            serverSocketChannel = ServerSocketChannel.open();// serverSocket.getChannel();
            outChannel = FileChannel.open(Paths.get("d:\apiserver.txt"), StandardOpenOption.WRITE, StandardOpenOption.CREATE);
            // 2.连接
            serverSocketChannel.bind(new InetSocketAddress(2653));
            // 3.获取客户端连接通道
            socketChannel = serverSocketChannel.accept();
            System.out.println("accept client success");
            // 4.分配缓冲区
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            // 5.接收客户端数据保存至本地
            while (socketChannel.read(byteBuffer)!=-1) {                
                byteBuffer.flip();
                outChannel.write(byteBuffer);
                byteBuffer.clear();
            }
            System.out.println("socket channel to server local channel");
            // 6.给客户端发送返回信息
            byteBuffer.put("file apiserver.txt has bean saved...".getBytes());
            byteBuffer.flip();// 切换到读模式
            socketChannel.write(byteBuffer);
            
        } catch (IOException ex) {
            ex.printStackTrace();
        }finally{
            if (serverSocketChannel!=null) {
                try {
                    serverSocketChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
            if (socketChannel!=null) {
                try {
                    socketChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
            if (outChannel!=null) {
                try {
                    outChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }

 打印结果：

accept client success
client local channel to socket channel
client wait for server's message...
socket channel to server local channel
client receive server msg:file apiserver.txt has bean saved...

非阻塞

　　Java NIO 是非阻塞模式的。

　　当线程从某通道进行读写数据时，若没有数据可用时，该线程可以进行其他任务。线程通常将非阻塞 IO 的空闲时间用于在其他通道上执行 IO 操作，所以单独的线程可以管理多个输入和输出通道。

　　因此， NIO 可以让服务器端使用一个或有限几个线程来同时处理连接到服务器端的所有客户端。

如何形成非阻塞IO（原理）

从上边的图中我们知道要构成NIO非阻塞模式，必须要引入Selector。那么，什么是Selector?

选择器（Selector）

选择器（Selector）是SelectableChannle对象的多路复用器，Selector可以同时监控多个SelectableChannel的IO状况，也就是说，利用Selector可以一个单独的线程管理多个Channel。Selector是非阻塞IO的核心。 

使用Selector的好处

使用更少的线程来就可以来处理通道了， 相比使用多个线程，避免了线程上下文切换带来的开销。

Selector（选择器）的使用方法介绍

1. Selector的创建

通过调用Selector.open()方法创建一个Selector对象，如下：

Selector selector = Selector.open();

这里需要说明一下

2. 注册Channel到Selector

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

Channel必须是非阻塞的。 
所以FileChannel不适用Selector，因为FileChannel不能切换为非阻塞模式，更准确的来说是因为FileChannel没有继承SelectableChannel。SocketChannel可以正常使用。

SelectableChannel抽象类 有一个 configureBlocking（） 方法用于使通道处于阻塞模式或非阻塞模式。

abstract SelectableChannel configureBlocking(boolean block)  

注意：

SelectableChannel抽象类的configureBlocking（） 方法是由 AbstractSelectableChannel抽象类实现的，SocketChannel、ServerSocketChannel、DatagramChannel都是直接继承了 AbstractSelectableChannel抽象类 。 
大家有兴趣可以看看NIO的源码，各种抽象类和抽象类上层的抽象类。我本人暂时不准备研究NIO源码，因为还有很多事情要做，需要研究的同学可以自行看看。

register() 方法的第二个参数。这是一个“ interest集合 ”，意思是在通过Selector监听Channel时对什么事件感兴趣。可以监听四种不同类型的事件：

• Connect
• Accept
• Read
• Write

通道触发了一个事件意思是该事件已经就绪。

这四种事件用SelectionKey的四个常量来表示：

SelectionKey.OP_CONNECT，某个Channel成功连接到另一个服务器称为“ 连接就绪 ”
SelectionKey.OP_ACCEPT，一个ServerSocketChannel准备好接收新进入的连接称为“ 接收就绪 ”
SelectionKey.OP_READ，一个有数据可读的通道可以说是“ 读就绪 ”
SelectionKey.OP_WRITE，等待写数据的通道可以说是“ 写就绪 ”

如果你对不止一种事件感兴趣，使用或运算符即可，如下：

int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;

3. SelectionKey介绍

一个SelectionKey键表示了一个特定的通道对象和一个特定的选择器对象之间的注册关系。

key.attachment(); //返回SelectionKey的attachment，attachment可以在注册channel的时候指定。
key.channel(); // 返回该SelectionKey对应的channel。
key.selector(); // 返回该SelectionKey对应的Selector。
key.interestOps(); //返回代表需要Selector监控的IO操作的bit mask
key.readyOps(); // 返回一个bit mask，代表在相应channel上可以进行的IO操作。

key.interestOps()

我们可以通过以下方法来判断Selector是否对Channel的某种事件感兴趣

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

key.readyOps()

ready 集合是通道已经准备就绪的操作的集合。JAVA中定义以下几个方法用来检查这些操作是否就绪.

//创建ready集合的方法
int readySet = selectionKey.readyOps();
//检查这些操作是否就绪的方法
key.isAcceptable();//是否可读，是返回 true
boolean isWritable()：//是否可写，是返回 true
boolean isConnectable()：//是否可连接，是返回 true
boolean isAcceptable()：//是否可接收，是返回 true

从SelectionKey访问Channel和Selector很简单。如下：

Channel channel = key.channel();
Selector selector = key.selector();
key.attachment();

可以将一个对象或者更多信息附着到SelectionKey上，这样就能方便的识别某个给定的通道。例如，可以附加与通道一起使用的Buffer，或是包含聚集数据的某个对象。使用方法如下：

key.attach(theObject);
Object attachedObj = key.attachment();

还可以在用register()方法向Selector注册Channel的时候附加对象。如：

SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

4. 从Selector中选择channel(Selecting Channels via a Selector)

选择器维护注册过的通道的集合，并且这种注册关系都被封装在SelectionKey当中.

Selector维护的三种类型SelectionKey集合

• 已注册的键的集合(Registered key set)

  所有与选择器关联的通道所生成的键的集合称为已经注册的键的集合。并不是所有注册过的键都仍然有效。这个集合通过 keys() 方法返回，并且可能是空的。这个已注册的键的集合不是可以直接修改的；试图这么做的话将引发java.lang.UnsupportedOperationException。

• 已选择的键的集合(Selected key set)

  所有与选择器关联的通道所生成的键的集合称为已经注册的键的集合。并不是所有注册过的键都仍然有效。这个集合通过 keys() 方法返回，并且可能是空的。这个已注册的键的集合不是可以直接修改的；试图这么做的话将引发java.lang.UnsupportedOperationException。

• 已取消的键的集合(Cancelled key set)

  已注册的键的集合的子集，这个集合包含了 cancel() 方法被调用过的键(这个键已经被无效化)，但它们还没有被注销。这个集合是选择器对象的私有成员，因而无法直接访问。

  注意： 
  当键被取消（ 可以通过isValid( ) 方法来判断）时，它将被放在相关的选择器的已取消的键的集合里。注册不会立即被取消，但键会立即失效。当再次调用 select( ) 方法时（或者一个正在进行的select()调用结束时），已取消的键的集合中的被取消的键将被清理掉，并且相应的注销也将完成。通道会被注销，而新的SelectionKey将被返回。当通道关闭时，所有相关的键会自动取消（记住，一个通道可以被注册到多个选择器上）。当选择器关闭时，所有被注册到该选择器的通道都将被注销，并且相关的键将立即被无效化（取消）。一旦键被无效化，调用它的与选择相关的方法就将抛出CancelledKeyException。

select()方法介绍

在刚初始化的Selector对象中，这三个集合都是空的。 通过Selector的select（）方法可以选择已经准备就绪的通道 （这些通道包含你感兴趣的的事件）。比如你对读就绪的通道感兴趣，那么select（）方法就会返回读事件已经就绪的那些通道。下面是Selector几个重载的select()方法：

• int select()：阻塞到至少有一个通道在你注册的事件上就绪了。
• int select(long timeout)：和select()一样，但最长阻塞时间为timeout毫秒。
• int selectNow()：非阻塞，只要有通道就绪就立刻返回。

select()方法返回的int值表示有多少通道已经就绪,是自上次调用select()方法后有多少通道变成就绪状态。之前在select（）调用时进入就绪的通道不会在本次调用中被记入，而在前一次select（）调用进入就绪但现在已经不在处于就绪的通道也不会被记入例如：首次调用select()方法，如果有一个通道变成就绪状态，返回了1，若再次调用select()方法，如果另一个通道就绪了，它会再次返回1。如果对第一个就绪的channel没有做任何操作，现在就有两个就绪的通道，但在每次select()方法调用之间，只有一个通道就绪了。

一旦调用select()方法，并且返回值不为0时，则 可以通过调用Selector的selectedKeys()方法来访问已选择键集合 。如下： 
Set selectedKeys=selector.selectedKeys(); 
进而可以放到和某SelectionKey关联的Selector和Channel。如下所示：

Set selectedKeys = selector.selectedKeys();
Iterator keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()) {
    SelectionKey key = keyIterator.next();
    if(key.isAcceptable()) {
        // a connection was accepted by a ServerSocketChannel.
    } else if (key.isConnectable()) {
        // a connection was established with a remote server.
    } else if (key.isReadable()) {
        // a channel is ready for reading
    } else if (key.isWritable()) {
        // a channel is ready for writing
    }
    keyIterator.remove();
}

5. 停止选择的方法

选择器执行选择的过程，系统底层会依次询问每个通道是否已经就绪，这个过程可能会造成调用线程进入阻塞状态,那么我们有以下三种方式可以唤醒在select（）方法中阻塞的线程。

• wakeup()方法 ：通过调用Selector对象的wakeup（）方法让处在阻塞状态的select()方法立刻返回 
  该方法使得选择器上的第一个还没有返回的选择操作立即返回。如果当前没有进行中的选择操作，那么下一次对select()方法的一次调用将立即返回。
• close()方法 ：通过close（）方法关闭Selector， 
  该方法使得任何一个在选择操作中阻塞的线程都被唤醒（类似wakeup（）），同时使得注册到该Selector的所有Channel被注销，所有的键将被取消，但是Channel本身并不会关闭。

使用NIO实现网络通信的三个核心

1、通道（channel）:负责连接

java.nio.channels.Channel接口：
　　|--SelectableChannel 
　　　　|--SocketChannel 
　　　　|--ServerSocketChannel
　　　　|--DatagramChannel

　　　　|--Pipe.SinkChannel 
　　　　|--Pipe.SourceChannel 

2、缓冲区（Buffer）:负责数据的存储

3、选择器（Selector）:是SelectableChannel的多路复用器。用于监控SelectableChannel的IO状况。

非阻塞IO示例

    public static void main(String[] args){
        new Thread(new Runnable() {
            @Override
            public void run() {
                server();
            }
        }).start();
        
        new Thread(new Runnable() {
            @Override
            public void run() {
                client();
            }
        }).start();
    }
    
    //static Charset charset = Charset.forName("UTF-8");
   // static CharsetEncoder encoder = charset.newEncoder();
   // static CharsetDecoder decoder = charset.newDecoder();
    
    // 非阻塞io,客户端
    public static void client(){
        
        SocketChannel socketChannel = null;// 网络channel
        try {
            // 1.获取通道
            socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 2653));
            // 2.切换为非阻塞模式
            socketChannel.configureBlocking(false);
            // 3.分配缓冲区
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            
            // 4.控制台输入的数据通过缓冲区传输至网络channel
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNext()) {
                byteBuffer.put(scanner.next().getBytes());// 数据放入缓冲区
                byteBuffer.flip();// 切换写模式
                socketChannel.write(byteBuffer);
                byteBuffer.clear();
            }
            System.out.println("client scanner info to socket channel");
            
            // 5.切换到接收数据的模式
            socketChannel.shutdownOutput();
            System.out.println("client wait for server's message...");
            
            // 6.接收服务端的数据
            int len;
            while ((len=socketChannel.read(byteBuffer))!=-1) {               
                byteBuffer.flip();
                System.out.println("client receive server msg:"+new String(byteBuffer.array(), 0, len));
                byteBuffer.clear();
            }
            
        } catch (IOException ex) {
            ex.printStackTrace();
        } finally{
            if (socketChannel!=null) {
                try {
                    socketChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
    // 非阻塞io,服务端
    public static void server(){
        ServerSocketChannel serverSocketChannel = null;// 网络channel
//        SocketChannel socketChannel = null;// 接收客户端的channel
        try {
            // 1.获取通道
            serverSocketChannel = ServerSocketChannel.open();// serverSocket.getChannel();
            // 2.设置非阻塞模式
            serverSocketChannel.configureBlocking(false);
            // 3.绑定连接
            serverSocketChannel.bind(new InetSocketAddress(2653));
            // 4.获取选择器
            Selector selector = Selector.open();
            // 5.将通道注册到选择器上,并且指定监听事件
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            // 6.循环获取选择器上已经准备就绪的事件
            while (selector.select()>0) {                
                // 7.获取当前选择器上所有注册的选择键（已就绪的监听事件）
                Iterator<SelectionKey> skeys = selector.selectedKeys().iterator();
                while (skeys.hasNext()) {
                    // 8.获取已经准备就绪的事件
                    SelectionKey skey = skeys.next();
                    // 9.判断这个事件是什么事件
                    if (skey.isAcceptable()) {// 接收就绪事件
                        System.out.println("server is acceptable...");
                        // 10.获取客户端连接
                        SocketChannel socketChannel = serverSocketChannel.accept();
                        // 11.置为非阻塞模式
                        socketChannel.configureBlocking(false);
                        // 12.将该通道注册到selector上
                        socketChannel.register(selector, SelectionKey.OP_READ);
                    } else if (skey.isReadable()) {// 读就绪事件
                        System.out.println("server is readable...");
                        // 10.获取当前选择器上的通道
                        SocketChannel selectableChannel = (SocketChannel)skey.channel();
                        // 11.准备读数据，先分配缓冲区
                        ByteBuffer bBuffer = ByteBuffer.allocate(1024);
                        int len_read = 0;
                        StringBuffer receivestr = new StringBuffer("server received:");
                        // 12.开始读
                        while ((len_read=selectableChannel.read(bBuffer))>0) {                            
                            bBuffer.flip();
                            receivestr.append(new String(bBuffer.array(),0,len_read));
                            bBuffer.clear();
                        }
                        System.out.println(receivestr.toString());
                    }if (skey.isValid() && skey.isWritable()) {// 判断是否有效及可以发送给客户端
                        System.out.println("server is writeable...");
                        // 客户端可写时
                        // 10.获取当前选择器上的通道
                        SocketChannel selectableChannel = (SocketChannel)skey.channel();
                        // 11.准备写数据，先分配缓冲区
                        ByteBuffer bBuffer = ByteBuffer.allocate(1024);
                        bBuffer.put("msg has received".getBytes());
                        selectableChannel.write(bBuffer);
                    }else {
                        System.out.println("server is other able...");
                    }
                    // 13.取消选择键SelectionKey
                    skeys.remove();
                }
                
            }
            
        } catch (IOException ex) {
            ex.printStackTrace();
        }finally{
            if (serverSocketChannel!=null) {
                try {
                    serverSocketChannel.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }