JDK AIO

jdk7中新增了一些与文件(网络)I/O相关的一些api。这些API被称为NIO.2,或称为AIO(Asynchronous I/O)。

AIO最大的一个特性就是异步能力,这种能力对socket与文件I/O都起作用。

实现方式

  • Future 方式

即提交一个 I/O 操作请求(accept/read/write),返回一个 Future。

然后您可以对 Future 进行检查(调用get(timeout)),确定它是否完成,或者阻塞 IO 操作直到操作正常完成或者超时异常。

使用 Future 方式很简单,需要注意的是,因为Future.get()是同步的,所以如果不仔细考虑使用场合,使用 Future 方式可能很容易进入完全同步的编程模式,从而使得异步操作成为一个摆设。

如果这样,那么原来旧版本的 Socket API 便可以完全胜任,大可不必使用异步 I/O.

  • Callback 方式

即提交一个 I/O 操作请求,并且指定一个 CompletionHandler。

当异步 I/O 操作完成时,便发送一个通知,此时这个 CompletionHandler 对象的 completed 或者 failed 方法将会被调用。

性能

因为AIO的实施需充分调用OS参与,IO需要操作系统支持、并发也同样需要操作系统的支持,所以性能方面不同操作系统差异会比较明显。

Future 实现方式

Server

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class AioFutureServer {

    private static final int DEFAULT_PORT = 12345;

    private AsynchronousServerSocketChannel serverSocketChannel;

    public AioFutureServer() throws IOException {
        serverSocketChannel = AsynchronousServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(DEFAULT_PORT));
        System.out.println("Server listen on port: " + DEFAULT_PORT);
    }

    public void startWithFuture() throws InterruptedException,
            ExecutionException, TimeoutException {
        while (true) {
            // 循环接收客户端请求
            Future<AsynchronousSocketChannel> future = serverSocketChannel.accept();
            // get() 是为了确保 accept 到一个连接
            AsynchronousSocketChannel socket = future.get();
            handleWithFuture(socket);
        }
    }

    /**
     * 处理未来的信息
     * @param channel 异步客户端
     */
    private void handleWithFuture(AsynchronousSocketChannel channel) throws InterruptedException, ExecutionException, TimeoutException {
        ByteBuffer readBuf = ByteBuffer.allocate(8);
        readBuf.clear();

        // 一次可能读不完
        while (true) {
            //get 是为了确保 read 完成,超时时间可以有效避免DOS攻击,如果客户端一直不发送数据,则进行超时处理
            Integer integer = channel.read(readBuf).get(10, TimeUnit.SECONDS);
            System.out.println("read: " + integer);
            if (integer == -1) {
                break;
            }
            readBuf.flip();
            System.out.println("received: " + Charset.forName("UTF-8").decode(readBuf));
            readBuf.clear();
        }
    }


    public static void main(String[] args) throws IOException, InterruptedException, ExecutionException, TimeoutException {
        new AioFutureServer().startWithFuture();
    }

}

客户端

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class AioClient {

    private static final int DEFAULT_PORT = 12345;

    public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
        AsynchronousSocketChannel client = AsynchronousSocketChannel.open();
        client.connect(new InetSocketAddress("localhost", DEFAULT_PORT)).get();
        client.write(ByteBuffer.wrap("123456789".getBytes()));
    }

}

测试

  • 启动服务端
Server listen on port: 12345
  • 启动客户端

服务端日志

read: 8
received: 12345678
read: 1
received: 9
Exception in thread "main" java.util.concurrent.ExecutionException: java.io.IOException: 指定的网络名不再可用。

Callback 模式

服务端

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;
import java.util.concurrent.TimeUnit;

public class AioCompletionServer {

    private static final int DEFAULT_PORT = 12345;

    private AsynchronousServerSocketChannel serverSocketChannel;

    public AioCompletionServer() throws IOException {
        serverSocketChannel = AsynchronousServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(DEFAULT_PORT));
        System.out.println("Server listen on port: " + DEFAULT_PORT);
    }

    /**
     * 使用回调的方式
     */
    public void startWithCompletionHandler() {
        serverSocketChannel.accept(null,
                new CompletionHandler<AsynchronousSocketChannel, Object>() {
                    @Override
                    public void completed(AsynchronousSocketChannel result, Object attachment) {
                        // 再此接收客户端连接
                        serverSocketChannel.accept(null, this);
                        // 处理结果
                        handleWithCompletionHandler(result);
                    }

                    @Override
                    public void failed(Throwable exc, Object attachment) {
                        exc.printStackTrace();
                    }
                });
    }

    /**
     * 处理异步的结果
     * @param channel 客户端信道
     */
    private void handleWithCompletionHandler(final AsynchronousSocketChannel channel) {
        try {
            final long timeout = 10L;
            final ByteBuffer buffer = ByteBuffer.allocate(8);

            // 再次读取,还是一种回调的方式。
            channel.read(buffer, timeout, TimeUnit.SECONDS, null, new CompletionHandler<Integer, Object>() {
                @Override
                public void completed(Integer result, Object attachment) {
                    System.out.println("read:" + result);
                    if (result == -1) {
                        try {
                            channel.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        return;
                    }
                    buffer.flip();
                    System.out.println("received message:" + Charset.forName("UTF-8").decode(buffer));
                    buffer.clear();

                    // 递归调用,直到结束为止。
                    channel.read(buffer, timeout, TimeUnit.SECONDS, null, this);
                }

                @Override
                public void failed(Throwable exc, Object attachment) {
                    exc.printStackTrace();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        new AioCompletionServer().startWithCompletionHandler();
        // 沉睡等待处理。
        TimeUnit.SECONDS.sleep(100);
    }

}

客户端

同上

参考资料

《Netty 权威指南》

Netty-07-通讯模型之 AIO

AIO 代码实现