Java IO编程全解(四)——NIO编程
阅读原文时间:2024年07月03日阅读:1

  转载请注明出处:http://www.cnblogs.com/Joanna-Yan/p/7793964.html

  前面讲到:Java IO编程全解(三)——伪异步IO编程

  NIO,即New I/O,这是官方叫法,因为它相对于之前的I/O类库是新增的。但是,由于之前老的I/O类库是阻塞I/O,New I/O类库的目标就是要让Java支持非阻塞I/O,所以,更多的人喜欢称之为非阻塞I/O(Non-block I/O),由于非阻塞I/O更能够体现NIO的特点,所以这里使用的NIO都是指非阻塞I/O。

  与Socket类和ServerSocket类相对应,NIO也提供了SocketChannel和ServerSocketChannel两种不同的套接字通道实现。这两种新增的通道都支持阻塞和非阻塞两种模式。阻塞模式使用非常简单,但是性能和可靠性都不好,非阻塞则正好相反。开发人员一般可以根据自己的需要来选择合适的模式,一般来说,低负载、低并发的应用程序可以选择同步阻塞I/O以降低编程复杂度,但是对于高负载、高并发的网络应用,需要使用NIO的非阻塞模式进行开发。

  新的输入/输出(NIO)库是在JDK1.4中引入的。NIO弥补了原来同步阻塞I/O的不足,它在标准Java代码中提供了高速的、面向块的I/O。通过定义包含数据的类,以及通过以块的形式处理这些数据,NIO不使用本机代码就可以利用低级优化,这是原来的I/O包所无法做到的。下面对NIO的一些概念和功能做下简单介绍,以便大家能够快速地了解NIO类库和相关概念。

  1.缓冲区Buffer

Buffer是一个对象,它包含一些要写入或者要读出的数据。在NIO类库中加入Buffer对象,体现了新库与原I/O的一个重要区别。在面向流的I/O中,可以将数据直接写入或者将数据直接读到Stream对象中。

  在NIO库中,所有数据都是用缓冲区处理的。在读取数据时,它是直接读到缓冲区中的;在写入数据时,写入到缓冲区中。任何时候访问NIO中的数据,都是通过缓冲区进行操作。

  缓冲区实质上是一个数组。通常它是一个字节数组(ByteBuffer),也可以使用其他种类的数组。但是缓冲区不仅仅是一个数组,缓冲区提供了对数据的结构化访问以及维护读写位置(limit)等信息。

  最常用的缓冲区是ByteBuffer,一个ByteBuffer提供了一组功能用于操作byte数组。除了ByteBuffer,还有其他的一些缓冲区,事实上,每一种Java基本类型(除了Boolean类型)都对应有一种缓冲区,具体如下:

  • ByteBuffer:字节缓冲区

  • CharBuffer:字符缓冲区

  • ShortBuffer:短整型缓冲区

  • IntBuffer:整型缓冲区

  • LongBuffer:长整型缓冲区

  • FloatBuffer:浮点型缓冲区

  • DoubleBuffer:双精度浮点型缓冲区

       每一个Buffer类都是Buffer接口的一个子实例。除了ByteBuffer,每一个Buffer类都有完全一样的操作,只是它们所处理的数据类型不一样。因为大多数标准I/O操作都是使用ByteBuffer,所以它除了具有一般缓冲区的操作之外还提供一些特有的操作,方便网络读写。

      2.通道Channel

    Channel是一个通道,可以通过它读取和写入数据,它就像自来水管一样,网络数据通过Channel读取和写入。通道与流的不同之处在于通道是双向的,流只是在一个方向上移动(一个流必须是InputStream或者OutputStream的子类),而且通道可以用于读、写或者同时读写。因为Channel是全双工的,所以它可以比流更好地映射底层操作系统的API。

      3.多路复用器Selector

      多路复用器Selector是Java NIO编程的基础,熟练地掌握Selector对于掌握NIO编程至关重要。多路复用器提供选择已经就绪的任务的能力。简单来讲,Selector会不断地轮询注册在其上的Channel,如果某个Channel上面有新的TCP连接接入、读和写事件,这个Channel就处于就绪状态,会被Selector轮询出来,然后通过SelectionKey可以获取就绪Channel的集合,进行后续的I/O操作。

      一个多路复用器Selector可以同时轮询多个Channel,由于JDK使用了epoll()代替传统的select实现,所以它并没有最大连接句柄1024/2048的限制。这也就意味着只需要一个线程负责Selector的轮询,就可以接入成千上万的客户端,这确实是个非常巨大的进步。

      NIO服务端通信序列图如下图所示:

  下面,我们对NIO服务端的主要创建过程进行讲解和说明,作为NIO的基础入门,我们将忽略掉一些在生产环境中部署所需要的一些特性和功能。

  步骤一:打开ServerSocketChannel,用于监听客户端的连接,它是所有客户端连接的父管道。

ServerSocketChannel acceptorSvr=ServerSocketChannel.open();

  步骤二:绑定监听端口,设置连接为非阻塞模式。

acceptorSvr.socket().bind(new InetSocketAddress(InetAddress.getByName("IP"),port));
acceptorSvr.configureBlocking(false);

  步骤三:创建Reactor线程,创建多路复用器并启动线程。

Selector selector=Selector.open();
New Thread(new ReactorTask()).start();

  步骤四:将ServerSocketChannel注册到Reactor线程的多路复用器Selector上,监听ACCEPT事件。

SelectionKey key=acceptorSvr.register(selector,SelectionKey.OP_ACCEPT,ioHandler);

  步骤五:多路复用器在线程run方法的无线循环体内轮询准备就绪的Key。

int num=selector.select();
Set selectedKeys=selector.selectedKeys();
Iterator it=selectedKeys.iterator();
while(it.hasNext()){
  SelectionKey key=(SelectionKey )it.next();
  //…deal with I/O event…
}

  步骤六:多路复用器监听到有新的客户端接入,处理新的接入请求,完成TCP三次握手,建立物理链路。

SocketChannel channel=svrChannel.accpet();

  步骤七:设置客户端链路为非阻塞模式。

channel.configureBlocking(false);
channel.socket().setReuseAddress(true);
……

  步骤八:将新接入的客户端连接注册到Reactor线程的多路复用器,监听读操作,用来读取客户端发送的网络消息。

SelectionKey key=socketChannel.register(selector,SelectionKey.OP_READ,ioHandler);

  步骤九:异步读取客户端请求消息到缓冲区。

int readNumber=channel.read(receivedBuffer);

  步骤十:对ByteBuffer进行编解码,如果有半包消息指针reset,继续读取后续的报文,将解码成功的消息封装成Task,投递到业务线程池中,进行业务逻辑编排。

Object message=null;
while(buffer.hasRemain()){
  byteBuffer.mark();
  Object message=decode(byteBuffer);
  if(message==null){
    byteBuffer.reset();
    break;
  }
  messageList.add(message);
}
if(!byteBuffer.hasRemain()){
  byteBuffer.clear();
}else{
  byteBuffer.compact();
}
if(messageList!=null& !messageList.isEmpty()){
  for(Object messageE: messageList){
    handlerTask(messageE);
  }
}

  步骤十一:将POJO对象encode成ByteBuffer,调用SocketChannel的异步write接口,将消息异步发送给客户端。

socketChannel.write(buffer);

  注意:如果发送区TCP缓冲区满,会导致写半包,此时,需要注册监听写操作位,循环写,直到整包消息写入TCP缓冲区。

  当我们了解创建NIO服务端的基本步骤之后,下面我们将前面的时间服务器程序通过NIO重写一遍,让大家能够学习到完整版的NIO服务端创建。

package joanna.yan.nio;

public class TimeServer {

public static void main(String\[\] args) {  
    int port=9090;  
    if(args!=null&&args.length>0){  
        try {  
            port=Integer.valueOf(args\[0\]);  
        } catch (Exception e) {  
            // 采用默认值  
        }  
    }

    MultiplexerTimeServer timeServer=new MultiplexerTimeServer(port);  
    new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();  
}  

}

package joanna.yan.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
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.util.Date;
import java.util.Iterator;
import java.util.Set;
/**
* 多路复用类
* 它是一个独立的线程,负责轮询多路复器Selector,可以处理多个客户端的并发接入。
* @author Joanna.Yan
* @date 2017年11月6日下午3:51:41
*/
public class MultiplexerTimeServer implements Runnable{

    private Selector selector;//多路复用器  
    private ServerSocketChannel servChannel;  
    private volatile boolean stop;

    /\*\*  
     \* 初始化多路复用器、绑定监听端口  
     \* @param port  
     \*/  
    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 stop(){
this.stop=true;
}
    @Override
public void run() {
while(!stop){
try {
//设置selector的休眠时间为1s,无论是否有读写等事件发生,selector每隔1s都被唤醒一次。
selector.select(1000);
//当有处于就绪状态的Channel时,selector就返回就绪状态的Channel的SelectionKey集合。
Set selectedKeys=selector.selectedKeys();
Iterator it=selectedKeys.iterator();
SelectionKey key=null;
//通过对就绪状态的Channel集合进行迭代,可以进行网络的异步读写操作。
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 (IOException e) {
e.printStackTrace();
}
}
       /*
* 多路复用器关闭后,所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭,所以不需要重复释放资源。
*/
if(selector!=null){
try {
selector.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
    private void handleInput(SelectionKey key) throws IOException{
if(key.isValid()){
//处理新接入的请求消息
//通过SelectionKey的操作位进行判断即可获知网络事件类型
if(key.isAcceptable()){
//Accept the new connection
ServerSocketChannel ssc=(ServerSocketChannel) key.channel();
SocketChannel sc=ssc.accept();
//-----以上操作相当于完成了TCP的三次握手,TCP物理链路正式建立------

                //将新创建的SocketChannel设置为异步非阻塞,同时也可以对其TCP参数进行设置,例如TCP接收和发送缓冲区的大小等。  
                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();
//由于实现我们得知客户端发送的码流大小,作为例程,我们开辟一个1K的缓冲区
ByteBuffer readBuffer=ByteBuffer.allocate(1024);
//由于已经设置SocketChannel为异步非阻塞模式,因此它的read是非阻塞的。
int readBytes=sc.read(readBuffer);
/*
* readBytes>0 读到了字节,对字节进行编解码;
* readBytes=0 没有读取到字节,属于正常场景,忽略;
* readByte=-1 链路已经关闭,需要关闭SocketChannel,释放资源
*/
            if(readBytes>0){
//将缓冲区当前的limit设置为position,position设置为0,用于后续对缓冲区的读取操作。
readBuffer.flip();
//根据缓冲区可读的字节个数创建字节数组
byte[] bytes=new byte[readBuffer.remaining()];
//调用ByteBuffer的get操作将缓冲区可读的字节数组复制到新创建爱你的字节数组中
readBuffer.get(bytes);
String body=new String(bytes, "UTF-8");
System.out.println("The time server receive order: "+body);
//如果请求指令是"QUERY TIME ORDER"则把服务器的当前时间编码后返回给客户端
String currentTime="QUERY TIME ORDER".equalsIgnoreCase(body) ? new 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);
//调用ByteBuffer的put操作将字节数组复制到缓冲区
writeBuffer.put(bytes);
writeBuffer.flip();
channel.write(writeBuffer);

            /\*  
             \* 需要指出的是,由于SocketChannel是异步非阻塞的,它并不保证一次性能够把需要发送的字节数组发送完,  
             \* 此时会出现“写半包”问题,我们需要注册写操作,不断轮询Selector,将没有发送完毕的ByteBuffer发送完毕,  
             \* 可以通过ByteBuffer的hasRemaining()方法判断消息是否发送完成。  
             \* 此处仅仅是各简单的入门级例程,没有演示如何处理“写半包”场景,后面会说到。  
             \*/  
        }  
    }  

}

  NIO客户端创建序列图如图所示。

  步骤一:打开SocketChannel,绑定客户端本地地址(可选,默认系统会随机分配一个可用的本地地址)

SocketChannel clientChannel=SocketChannel.open();

  步骤二:设置SocketChannel为非阻塞模式,同时设置客户端连接的TCP参数。

clientChannel.configureBlocking(false);
socket.setReuseAddress(true);
socket.setReceiveBufferSize(BUFFER_SIZE);
socket.setSendBufferSize(BUFFER_SIZE);

  步骤三:异步连接服务端。

boolean connected=clientChannel.connect(new InetSocketAddress("ip",port));

  步骤四:判断是否连接成功,如果连接成功,则直接注册读状态位到多路复用器中,如果当前没有连接成功(异步连接,返回false,说明客户端已经发送sync包,服务端没有返回ack包,物理链路还没有建立)。

if(connected){
  clientChannel.register(selector,SelectionKey.OP_READ,ioHandler);
}else{
  clientChannel.register(selector,SelectionKey.OP_CONNECT,ioHandler);
}

  步骤五:向Reactor线程的多路复用器注册OP_CONNECT状态位,监听服务端的TCP ACK应答。

clientChannel.register(selector,SelectionKay.OP_CONNECT,ioHandler);

  步骤六:创建Reactor线程,创建多路复用器并启动线程。

Selector selector=Selector.open();
new Thread(new ReactorTask()).start();

  步骤七:多路复用器在线程run方法的无限循环体内轮询准备就绪的key。

int num=selector.select();
Set selectedKeys=selector.selectedKeys();
Iterator it=selectedKeys.iterator();
while(it.hasNext()){
  SelectionKey key=(SelectionKey)it.next();
  //…deal with I/O event…
}

  步骤八:接收connect事件进行处理。

if(key.isConnectable()){
  //handlerConnect();
}

  步骤九:判断连接结果,如果连接成功,注册读事件到多路复用器。

if(channel.finishConnect()){
  registerRead();
}

  步骤十:注册读事件到多路复用器。

clientChannel.register(selector,SelectionKey.OP_READ,ioHandler);

  步骤十一:异步读客户端请求消息到缓冲区。

int readNumber=channel.read(receivedBuffer);

  步骤十二:对ByteBuffer进行编解码,如果有半包消息接收缓冲区Reset,继续读取后续的报文,将解码成功的消息封装成Task,投递到业务线程池中,进行业务逻辑编排。

Object message=null;

while(buffer.hasRemain()){
  byteBuffer.mark();
  Object message=decode(byteBuffer);
  if(message==null){
    byteBuffer.reset();
    break;
  }
  messageList.add(message);
}

if(!byteBuffer.hasRemain()){
  byteBuffer.clear();
}else{
  byteBuffer.compact();
}

if(messageList!=null & !messageList.isEmpty()){
  for(Object messageE:messageList){
    handlerTask(messageE);
  }
}

  步骤十三:将POJO对象encode成ByteBuffer,调用SocketChannel的异步write接口,将消息异步发送给客户端。

socketChannel.wirte(buffer);

package joanna.yan.nio;

public class TimeClient {
public static void main(String[] args) {
int port=9090;
if(args!=null&&args.length>0){
try {
port=Integer.valueOf(args[0]);
} catch (Exception e) {
// 采用默认值
}
}

    new Thread(new TimeClientHandle("127.0.0.1", port),"TimClient-001").start();  
}  

}

package joanna.yan.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
/**
* 处理异步连接和读写操作
* @author Joanna.Yan
* @date 2017年11月6日下午4:33:14
*/
public class TimeClientHandle implements Runnable{
private String host;
private int port;
private Selector selector;
private SocketChannel socketChannel;
private volatile boolean stop;

/\*\*  
 \* 初始化NIO的多路复用器和SocketChannel对象  
 \* @param host  
 \* @param port  
 \*/  
    public TimeClientHandle(String host,int port){  
    this.host=host==null ? "127.0.0.1" : host;  
    this.port=port;  
    try {  
        selector=Selector.open();  
        socketChannel=SocketChannel.open();  
        //设置为异步非阻塞模式,同时还可以设置SocketChannel的TCP参数。例如接收和发送的TCP缓冲区大小  
        socketChannel.configureBlocking(false);  
    } catch (IOException e) {  
        e.printStackTrace();  
        System.exit(1);  
    }  
}  
@Override  
public void run() {  
    try {  
        doConnect();  
    } catch (IOException e) {  
        e.printStackTrace();  
        System.exit(1);  
    }  
    while(!stop){  
        try {  
            selector.select(1000);  
            Set<SelectionKey> selectedKeys=selector.selectedKeys();  
            Iterator<SelectionKey> it=selectedKeys.iterator();  
            SelectionKey key=null;  
            while(it.hasNext()){//轮询多路复用器Selector,当有就绪的Channel时  
                key=it.next();  
                it.remove();  
                try {  
                    handleInput(key);  
                } catch (Exception e) {  
                    if(key!=null){  
                        key.cancel();  
                        if(key.channel()!=null){  
                            key.channel().close();  
                        }  
                    }  
                }  
            }

        } catch (IOException e) {  
            e.printStackTrace();  
            System.exit(1);  
        }  
    }  
        //多路复用器关闭后,所有注册在上面的Channel和Pipe等资源都会被自动注册并关闭,所以不需要重复释放资源。  
    /\*  
     \* 由于多路复用器上可能注册成千上万的Channel或者pipe,如果一一对这些资源进行释放显然不合适。  
     \* 因此,JDK底层会自动释放所有跟此多路复用器关联的资源。  
     \*/

    if(selector!=null){  
        try {  
            selector.close();  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
    }  
}

    //多路复用器关闭后,所有注册在上面的Channel和Pipe等资源都会被自动注册并关闭,所以不需要重复释放资源。  
    /\*  
     \* 由于多路复用器上可能注册成千上万的Channel或者pipe,如果一一对这些资源进行释放显然不合适。  
     \* 因此,JDK底层会自动释放所有跟此多路复用器关联的资源。  
     \*/

    if(selector!=null){  
        try {  
            selector.close();  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
    }  
}  
    private void handleInput(SelectionKey key) throws ClosedChannelException, IOException {  
    if(key.isValid()){  
        //判断是否连接成功  
        SocketChannel sc=(SocketChannel) key.channel();  
        if(key.isConnectable()){//处于连接状态,说明服务器已经返回ACK应答消息  
            if(sc.finishConnect()){//对连接结果进行判断  
                /\*  
                 \* 将SocketChannel注册到多路复用器上,注册SelectionKey.OP\_READ操作位,  
                 \* 监听网络读操作,然后发送请求消息给服务端。  
                 \*/  
                sc.register(selector, SelectionKey.OP\_READ);  
                doWrite(sc);  
            }else{  
                System.exit(1);//连接失败,进程退出  
            }  
        }  
                    if(key.isReadable()){  
            //开辟缓冲区  
            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("Now is: "+body);  
                this.stop=true;  
            }else if(readBytes<0){  
                //对端链路关闭  
                key.cancel();  
                sc.close();  
            }else{  
                //读到0字节,忽略  
            }  
        }  
    }  
}  
    private void doConnect() throws IOException {  
    //如果直接连接成功,则将SocketChannel注册到多路复用器Selector上,发送请求消息,读应答  
    if(socketChannel.connect(new InetSocketAddress(host, port))){  
        socketChannel.register(selector, SelectionKey.OP\_READ);  
        doWrite(socketChannel);  
    }else{  
        /\*  
         \* 如果没有直接连接成功,则说明服务端没有返回TCP握手应答信息,但这并不代表连接失败,  
         \* 我们需要将SocketChannel注册到多路复用器Selector上,注册SelectionKey.OP\_CONNECT,  
         \* 当服务端返回TCP syn-ack消息后,Selector就能轮询到整个SocketChannel处于连接就绪状态。  
         \*/  
        socketChannel.register(selector, SelectionKey.OP\_CONNECT);  
    }  
}

    private void doWrite(SocketChannel sc) throws IOException {  
    byte\[\] req="QUERY TIME ORDER".getBytes();  
    ByteBuffer writeBuffer=ByteBuffer.allocate(req.length);  
    //写入到发送缓冲区中  
    writeBuffer.put(req);  
    writeBuffer.flip();  
    //由于发送是异步的,所以会存在"半包写"问题,此处不赘述  
    sc.write(writeBuffer);  
    if(!writeBuffer.hasRemaining()){//如果缓冲区中的消息全部发送完成  
        System.out.println("Send order 2 server succeed.");  
    }  
}  

}

  通过源码对比分析发现,NIO编程难度确实比同步阻塞BIO大很多,此处我们的NIO例程并没有考虑“半包读”和“半包写”,如果加上这些,代码会更加复杂。NIO代码既然这么复杂,为什么它的应用却越来越广泛呢,使用NIO编程的优点总结如下:

  1. 客户端发起的连接操作是异步的,可以通过多路复用器注册OP_CONNECT等待后续结果,不需要像之前的客户端那样被同步阻塞。

  2. SocketChannel的读写操作都是异步的,如果没有可读写的数据它不会同步等待,直接返回,这样I/O通信线程就可以处理其他的链路,不需要同步等待这个链路可用。

  3. 线程模型的优化:由于JDK的Selector在Linux等主流操作系统上通过epoll实现,它没有连接句柄数的限制(只受限于操作系统的最大句柄数或者对单个进程的句柄限制),这意味着一个Selector线程可以同时处理成千上万个客户端连接,而且性能不会随着客户端的增加而线性下降,因此,它非常适合做高性能、高负载的网络服务器。

      JDK1.7升级了NIO类库,升级后的NIO类库被称为NIO 2.0。引入注目的是,Java正式提供了异步文件I/O操作,同时提供了与UNIX网络编程事件驱动I/O对应的AIO。

Java IO编程全解(五)——AIO编程

如果此文对您有帮助,微信打赏我一下吧~