Java核心技术读书笔记11-6 Java AIO(NIO 2.0)

Posted 芝芝与梅梅

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Java核心技术读书笔记11-6 Java AIO(NIO 2.0)相关的知识,希望对你有一定的参考价值。

1.什么是Java AIO?

在Java SE 7 API中NIO相关库得到了扩充,新的NIO库被称为NIO 2.0,在这一版中Java提供了异步IO的能力,包括异步处理文件以及异步访问Socket数据。

异步的IO操作是IO模型的一种(关于IO模型,可以参考:Java NIO预备知识:I/O底层原理与网络I/O模型),它采用的是一种“订阅-通知”模式。即应用程序线程向操作系统注册IO监听,当应用程序线程发出IO请求后便可以不在关心IO处理,内核进行系统调用,当数据准备好时进行接收数据,接收完毕后还会自行将数据写入用户地址空间。写入完毕之后由内核主动通知应用程序。因此,AIO是异步非阻塞的。

对于Java的AIO来说,数据的读写操作直接调用API的read或write方法即可。
对于读操作来说,数据源中有数据可读,内核会将数据读入read方法的缓冲区(用户空间)然后通知应用程序。对于写操作来说,内核会将write方法传递到数据源,传递完毕后通知应用程序。两个方法都是异步的,这样的异步操作不是由发起I/O请求的线程来处理,而是通过线程池分配一个新的线程来进行处理。最后,对于异步操作的结果处理分为两种:返回一个Future对象来跟踪异步任务或注册时使用CompletionHandler在异步I/O完成后执行回调函数。
与第一版NIO的四种通道类似,Java SE 7为Java.nio.channels包新增了四种异步通道:
AsynchronousFileChannel
AsynchronousServerSocketChannel
AsynchronousSocketChannel
AsynchronousDatagramChannel
分别对应文件、TCP请求、TCP连接以及UDP的数据传输。

2.异步执行流程与异步通道组

2.1 异步执行的详细流程

1.首先,Java的异步任务分为两种:进行异步的IO操作与为IO结果执行回调函数。
2.Java是通过异步通道组中的线程池为异步任务分配线程,这个线程组可以在创建四种异步通道时指定,如果不指定则使用系统默认的组。
3.在用户线程执行任务时,会将IO请求提交给操作系统内核,如果存在CompletionHandler则一同注册到内核,然后立即返回继续执行线程的其余代码。
4.内核会使用系统调用完成相应的IO操作,例如读操作为:从数据源(文件、Socket)读数据写入用户线程的缓冲区。
5.IO完毕之后内核指定操作系统合适的Proactor(完成分发器——completion dispatcher)完成分发任务,即执行CompletionHandler指定的回调函数。

若不在创建通道时指定组,则会使用Java虚拟机维护的一个自动构造的系统范围的默认组,默认组有一个关联的线程池,可以根据需要创建新线程。如果没有配置默认组的ThreadFactory,则该组的线程池中的线程为守护线程。这意味着,如果你在主线程中构建了一个使用未配置的默认组的通道,有可能主线程会在异步任务执行完毕前退出,而此时守护线程也将因为无其他线程运行而退出,该任务将立即终止。

2.2 异步通道组

异步通道组关联一个线程池,在创建时指定。随后可以将该组作为参数传入一个异步通道。使用一个组的通道共享该组的线程池。

//        ExecutorService executorService = Executors.newSingleThreadExecutor(); //一个单线程线程池
//        AsynchronousChannelGroup group = AsynchronousChannelGroup.withThreadPool(executorService);
//        AsynchronousChannelGroup group = AsynchronousChannelGroup.withCachedThreadPool(Executors.newCachedThreadPool(), 10);//缓存线程池
        AsynchronousChannelGroup group = AsynchronousChannelGroup.withFixedThreadPool(10, Executors.defaultThreadFactory());//固定线程池
        AsynchronousSocketChannel channel = AsynchronousSocketChannel.open(group); //建立通道并指定通道组

3.异步读写文件

和同步的NIO类似,AIO也需要使用buffer来读取数据。异步读写文件的方式分两种分别是使用Future对象和

3.1 使用Future对象接收异步任务

读数据
首先,需要使用AsynchronousFileChannel类的静态方法open来返回一个AsynchronousFileChannel对象,接下来是将管道连接至一个缓冲区并将数据读到缓冲区内。
与同步读写File的NIO不同,该方法不会返回读入了多少字节,而是会立即返回一个Future对象,该对象即为在多线程编程中遇到过的java.util.concurrent.Future,该类可以处理异步任务,并且可以使用isDone实例方法检查任务是否完毕以及get方法获取结果。下面为示例代码:

        AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get("file/Test.txt"), StandardOpenOption.WRITE, StandardOpenOption.READ); //获得一个异步文件管道
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        Future<Integer> future = channel.read(buffer, 0); //读入数据到缓冲区,第二个参数为下标,指示文件的开始读取位置
        while(!future.isDone()); //如果异步任务未完成则进行阻塞
        buffer.flip(); 
        FirstNIO.printBufByte(buffer, "AIOTestOne"); //一个自定义的打印缓冲区结果方法

读数据
写操作类似,写这个过程也是一个任务,同样返回Future对象并根据这个对象判断任务执行的状态。

        buffer.put("abcdef".getBytes());
        buffer.flip();
        Future<Integer> future = channel.write(buffer, channel.size()); //从缓冲区写出数据到文件,第二个参数为下标,指示文件的开始写出位置
        while (!future.isDone()) ; //如果异步任务未完成则进行阻塞
        System.out.println("写入完成");
3.2 使用CompletionHandler

读数据
第二种方式是使用CompletionHandler,需要构建一个CompletionHandler的实现类对象,然后将该对象作为参数传入read方法。CompletionHandler接口的两个抽象方法分别是completed与failed,这两个方法承担了回调方法的作用,在read完毕后如果成功会调用completed,失败则会调用failed方法。因此可以在实现接口时重写的completed方法内部编写接收数据的处理逻辑。
使用CompletionHandler后,本线程发起一个I/O请求,调用read方法将这个请求提交给内核,内核处理完毕之后JVM会再使用一个新的线程,然后会由这个新线程使用回调方法。

        AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get("file/Test.txt"), StandardOpenOption.WRITE, StandardOpenOption.READ); //获得一个异步文件管道
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        /**
         * 第一个参数buffer会接收数据,第二个参数指定文件起始操作位置,第三个参数attachment也可以接收数据并可传递给completed方法
         */
        channel.read(buffer, 0, buffer, new CompletionHandler<Integer, ByteBuffer>() { 
            @Override
            public void completed(Integer result, ByteBuffer attachment) { //第一个参数为读取字节数,第二个参数为attachment接收数据
                System.out.println(Thread.currentThread().getName()); //一个新线程
                System.out.println("读入了" + result + "个字节");
                attachment.flip();
                FirstNIO.printBufByte(attachment, "任务自动处理");
                attachment.clear();
            }

            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                System.out.println("读入数据失败");
            }
        });
        Thread.sleep(10); //防止本线程结束时IO线程还未创建而无法接受数据。

写数据
写出数据类似,可以在completed方法中查看写出的字节数与写出数据。

AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get("file/Test.txt"), StandardOpenOption.WRITE, StandardOpenOption.READ); //获得一个异步文件管道
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        buffer.put("\\nnewStr".getBytes());
        buffer.flip();
        /**
         * 第一个参数buffer会接收数据,第二个参数指定文件起始操作位置,第三个参数attachment也可以查看写出数据并可传递给completed方法
         * 下述IO操作都由一个新线程执行
         */
        channel.write(buffer, channel.size(), buffer, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer result, ByteBuffer attachment) { //第一个参数为写出字节数,第二个参数为attachment写出数据
                System.out.println(Thread.currentThread().getName()); //一个新线程
                System.out.println("写出了" + result + "个字节"); //写出字节数量
                attachment.flip(); //可以查看写出的字节
                FirstNIO.printBufByte(attachment, "任务自动处理");
                attachment.clear();
            }

            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                System.out.println("写出数据失败");
            }
        });
        Thread.sleep(100); //防止本线程结束时IO线程还未创建而无法写出数据。

4.异步建立TCP连接与读写Socket数据

异步建立TCP连接与读写Socket数据与使用多路复用NIO的步骤非常相似,使用的分别是AsynchronousServerSocketChannel与AsynchronousSocketChannel,前者监听端口并通过accept方法获取TCP连接,并返回连接所用Socket。后者的一端连接有TCP所用的Socket用以传输数据。

4.1 使用Future对象接收异步任务

使用Future对象接收异步任务,对于异步Channel的建立连接、读、写都会返回Future对象,如果需要确保I/O确实完成了,需要该对象的isDone方法返回true。下面为对应的客户端与服务器端的代码。

//客户端
public class AIOClient {
    public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        InetSocketAddress address = new InetSocketAddress("localhost", 8189);
        AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();
        Future<Void> connectFuture = channel.connect(address);
        while(!connectFuture.isDone()); //等待直到建立连接
        Future<Integer> writeFuture;
        Future<Integer> readFuture;
        if(connectFuture.get() == null){
            System.out.println("已成功连接服务器");
            readFuture = channel.read(buffer);
            while (!readFuture.isDone());
            while (true){
                buffer.flip();
                FirstNIO.printBufByte(buffer, "服务器端");
                Scanner scanner = new Scanner(System.in);
                String message = "";
                message = scanner.nextLine();
                buffer.put(message.getBytes());
                buffer.flip();
                writeFuture = channel.write(buffer);
                while(!writeFuture.isDone());
                buffer.clear();
                readFuture = channel.read(buffer);
                while (!readFuture.isDone());
            }
        }else{
            System.out.println("无法建立连接");
        }
    }
}

//服务器端
public class AIOServer {
    public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open();
        InetSocketAddress address = new InetSocketAddress(8189);
        server.bind(address);
        System.out.println("服务器启动成功,正在等待连接...");
        while (true){
            Future<AsynchronousSocketChannel> future = server.accept();
            while(!future.isDone());
            System.out.println("建立新连接成功");
            Future<Integer> writeFuture;
            Future<Integer> readFuture;
            AsynchronousSocketChannel channel = future.get();
            buffer.put("Hello, here is server, welcome!".getBytes());
            buffer.flip();
            writeFuture = channel.write(buffer);
            while(!writeFuture.isDone());
            buffer.clear();
            readFuture = channel.read(buffer);
            while (!readFuture.isDone());
            while (true){
                buffer.flip();
                FirstNIO.printBufByte(buffer, "客户端");
                buffer.put("message received".getBytes());
                buffer.flip();
                writeFuture = channel.write(buffer);
                while(!writeFuture.isDone());
                buffer.clear();
                readFuture = channel.read(buffer);
                while (!readFuture.isDone());
            }
        }
    }
}
4.2 使用CompletionHandler

与读写文件类似,异步的ServerSocketChannel与SocketChannel的连接、读、写方法都有指定CompletionHandler对象的版本,在操作成功时会使用completed方法处理,否则会交给failed方法处理。
需要注意的是,与异步FileChannel不同,网络异步操作的连接、读写的操作可能是不断的,所以为了处理不断的操作,应该在completed与failed方法内再调用一次相关的方法以便继续处理下次操纵。
下面是将服务器端的连接操作改造为CompletionHandler版本的示例代码,注意server.accept(null, this);

public class AIOServer2 {
    public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {

        AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open();
        InetSocketAddress address = new InetSocketAddress(8193);
        server.bind(address);
        System.out.println("服务器启动成功,正在等待连接...");
        server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
            ByteBuffer buffer = ByteBuffer.allocate(1024);

            @Override
            public void completed(AsynchronousSocketChannel result, Object attachment) {
                server.accept(null, this); //处理接续进入的连接
                System.out.println("建立新连接成功");
                Future<Integer> writeFuture;
                Future<Integer> readFuture;
                AsynchronousSocketChannel channel = result;
                buffer.put("Hello, here is server, welcome!".getBytes());
                buffer.flip();
                writeFuture = channel.write(buffer);
                while (!writeFuture.isDone()) ;
                buffer.clear();
                readFuture = channel.read(buffer);
                while (!readFuture.isDone()) ;
                while (true) {
                    buffer.flip();
                    FirstNIO.printBufByte(buffer, "客户端");
                    buffer.put("message received".getBytes());
                    buffer.flip();
                    writeFuture = channel.write(buffer);
                    while (!writeFuture.isDone()) ;
                    buffer.clear();
                    readFuture = channel.read(buffer);
                    while (!readFuture.isDone()) ;
                }
            }

            @Override
            public void failed(Throwable exc, Object attachment) {
                server.accept(null, this); //处理接续进入的连接
                System.out.println("建立连接失败");
                System.out.println(exc.getMessage());
            }
        });
        while (true){
            Thread.sleep(5000);
        }
    }
}

参考:
AIO 详解
17. Java NIO AsynchronousFileChannel异步文件通道
Proactor模式详解

以上是关于Java核心技术读书笔记11-6 Java AIO(NIO 2.0)的主要内容,如果未能解决你的问题,请参考以下文章

Java核心技术读书笔记11- 1 Java流与读写器

[读书笔记]java核心技术

Java核心技术读书笔记12-Java SE 8引入的流

读书笔记--Java核心技术--基础篇

读书笔记--Java核心技术--高级特征

《Java核心技术》 -- 读书笔记 ① - 预热