Netty学习之二——BIO、NIO、AIO

深藏阁楼爱情的钟 2022-02-22 13:52 257阅读 0赞

**目录**

BIO

BIO 通信模型图

伪异步I/O编程

伪异步I/O模型图

伪异步I/O弊端分析

NIO

NIO类库简介

1、缓冲区buffer

2、通道 Channel

3、多路复用器 Selector

NIO服务端序列图

NIO 客户端创建序列

AIO

总结：

不同 I/O 模型对比

--------------------

# BIO #

网络编程的基本模型是Client/Server 模型，也就是两个进程之间进行互相通信，其中服务端提供位置信息（绑定的IP地址和监听端口），客户端通过连接操作向服务器监听的地址发起连接请求，通过[三次握手][Link 1]建立连接，如果连接成功建立，双方就可以通过网络套接字（Socket）进行通信。

## BIO 通信模型图 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70][]

采用BIO通信模型的服务端，通常由一个独立的Acceptor线程负责监听客户端的连接，它接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理，处理完成之后，通过输出流返回应答给客户端，线程销毁。这就是典型的一请求一应答通信模型。

该模型最大的问题是缺乏弹性伸缩能力，当客户端并发访问量增加后，服务端的线程个数和客户端并发访问数成1:1的正比关系，由于线程是Java虚拟机非常宝贵的系统资源，当线程数膨胀之后，系统的性能将急剧下降，随着并发访问量的继续增大，系统会发生线程堆栈溢出、创建新线程失败等问题，并最终导致进程宕机或者僵死，不能对外提供服务。

代码如下：

package com.netty.bio;
    
    import java.io.IOException;
    import java.net.ServerSocket;
    import java.net.Socket;
    
    
    public class TimeServer {
        public static void main(String[] args) {
            int port = 8080;
            ServerSocket serverSocket = null;
            try {
                serverSocket = new ServerSocket(port);
                System.out.println("Server start in port: " + port);
                Socket socket = null;
                while (true) {
                    socket = serverSocket.accept();
                    new Thread(new TimeServerHandler(socket)).start();
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if (serverSocket != null) {
                    System.out.println("Server close");
                    try {
                        serverSocket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    serverSocket = null;
                }
            }
        }
    }

package com.netty.bio;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.io.PrintWriter;
    import java.net.Socket;
    import java.util.Date;
    
    
    public class TimeServerHandler implements Runnable{
        private Socket socket;
    
        public TimeServerHandler(Socket socket) {
            this.socket = socket;
        }
    
        @Override
        public void run() {
            BufferedReader reader = null;
            PrintWriter writer = null;
            try {
                reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                writer = new PrintWriter(this.socket.getOutputStream(), true);
                String now = null;
                String body = null;
                while (true) {
                    body = reader.readLine();
                    if (body == null) {
                        break;
                    }
                    System.out.println("Server recv order : " + body);
                    now = "QUERY TIME ORDER".equals(body) ? new Date(System.currentTimeMillis()).toString()
                            : "BAD ORDER";
                    writer.println(now);
                }
            } catch (IOException e) {
                e.printStackTrace();
    
            } finally {
                if (reader != null) {
                    try {
                        reader.close();
                        reader = null;
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                if (writer != null) {
                    writer.close();
                    writer = null;
                }
                if (socket != null) {
                    try {
                        socket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    socket = null;
                }
            }
        }
    }

package com.netty.bio;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.io.PrintWriter;
    import java.net.Socket;
    
    public class TimeClient {
        public static void main(String[] args) {
            int port = 8080;
            Socket socket = null;
            BufferedReader reader = null;
            PrintWriter writer = null;
            try {
                socket = new Socket("127.0.0.1", port);
                reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                writer = new PrintWriter(socket.getOutputStream(), true);
                writer.println("QUERY TIME ORDER");
                System.out.println("Send Order To Server Success");
                String resp = reader.readLine();
                System.out.println("Now is : " + resp);
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if (reader != null) {
                    try {
                        reader.close();
                        reader = null;
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                if (writer != null) {
                    writer.close();
                    writer = null;
                }
                if (socket != null) {
                    try {
                        socket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    socket = null;
                }
            }
        }
    }

# 伪异步I/O编程 #

为了解决同步阻塞I/O面临的一个链路需要一个线程处理的问题，后来有人对他的线程模型进行了优化——后端通过一个线程池来处理多个客户端的请求接入，形成客户端个数M：线程池最大线程数N的比例关系，其中M可以远远大于N。

## 伪异步I/O模型图 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70 1][]

当有新的客户端接入时，将客户端的Socket封装成一个Task投递到后端的线程池中进行处理，JDK的线程池维护一个消息队列和N个活跃线程，对消息队列中的任务进行处理。

package com.netty.bio;
    
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class TimeServerHandlerExecutePool {
        private ExecutorService executor;
    
        public TimeServerHandlerExecutePool(int maxPoolSize, int queueSize) {
            executor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
                    maxPoolSize, 120L, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<Runnable>(queueSize));
        }
        public void execute(Runnable task) {
            executor.execute(task);
        }
    }

package com.netty.bio;
    
    import java.io.IOException;
    import java.net.ServerSocket;
    import java.net.Socket;
    
    /**
     * Created by 18121254 on 2019/4/9.
     */
    public class TimeServer {
        public static void main(String[] args) {
            int port = 8080;
            ServerSocket serverSocket = null;
            try {
                serverSocket = new ServerSocket(port);
                System.out.println("Server start in port: " + port);
                Socket socket = null;
                TimeServerHandlerExecutePool pool = new TimeServerHandlerExecutePool(50, 10000);
                while (true) {
                    socket = serverSocket.accept();
    //                new Thread(new TimeServerHandler(socket)).start();
                    pool.execute(new TimeServerHandler(socket));
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if (serverSocket != null) {
                    System.out.println("Server close");
                    try {
                        serverSocket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    serverSocket = null;
                }
            }
        }
    }

## 伪异步I/O弊端分析 ##

InputStream的read方法，是个阻塞方法，当对Socket的输入流进行读取操作的时候，它会一致阻塞下去，知道发生如下三种事情。

1、有数据可读；

2、可用数据已经读取完毕；

3、发生空指针或者I/O异常。

这意味着当对方发送请求或者应答消息比较缓慢，或者网络传输较慢时，读取输入流一方的通信线程将被长时间阻塞，如果对方要60s才能够将数据发送完成，读取一方的I/O线程也将会被同步阻塞60s，在此期间，其他接入消息只能在消息队列中排队。

当调用 OutputStream 的 write方法写输出流的时候，它将会被阻塞，直到所有要发送的字节全部写入完毕，或者发生异常。学习过TCP/IP相关知识的人都知道，当消息的接收方处理缓慢的时候，将不能及时的从TCP缓冲区读取数据，这将会导致发送发的TCP window size不断减小，直到为0，双方处于 Keep-Alive状态，消息发送方将不能再向TCP缓冲区写入消息，这时如果采用的是同步阻塞I/O，write操作将会被无限期阻塞，直到TCP window size 大于 0 或者发生I/O异常。

伪异步I/O实际上仅仅是对之前I/O线程模型的一个简单优化，它无法从根本上解决同步I/O导致的通信线程阻塞问题。

未解决这个问题，引入了NIO。

# NIO #

Non-block I/O，非阻塞I/O

## NIO类库简介 ##

NIO是jdk1.4中引入的。

### 1、缓冲区buffer ###

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

缓冲区实质上是一个数组。通常它是一个字节数组（ByteBuffer），缓冲区提供了对数据的结构化访问以及维护读写位置等信息。每一种Java基本类型（除了Boolean）都对应一种缓冲区。

### 2、通道 Channel ###

网络数据通过 Channel 读取和写入。通道与流的不同之处在于通道是双向的，流只是在一个方向上移动，而通道可以用于读、写或者二者同时进行。

因为Channel 是全双工的，所以它可以比流更好地映射底层操作系统的 API。特别是在UNIX网络编程模型中，底层操作系统的通道都是全双工的，同时支持读写操作。

### 3、多路复用器 Selector ###

Selector是Java NIO编程的基础。多路复用器提供选择已经就绪的任务的能力。简单来讲，Selector会不断地轮询注册在其上的 Channel，如果某个 Channel 上面发生读或者写时间，这个Channel就处于就绪状态，会被Selector 轮询出来，然后通过 SelectionKey 可以获取就绪 Channel 的集合，进行后续的 I/O 操作。

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

### NIO 步骤 ###

1、创建 ServerSocketChannel，配置它为非阻塞模式；

2、绑定监听，配置 TCP参数，例如 backlog大小；

3、创建一个独立的 I/O 线程，用于轮询多路复用器 Selector；

4、创建 Selector，将之前创建的 ServerSocketChannel 注册到 Selector 上，监听 SelectionKey.ACCEPT；

5、启动 I/O 线程，在循环体中执行 Selector.select() 方法，轮询就绪的 Channel；

6、当轮询了处于就绪状态的 Channel时，需要对其进行判断，如果是 OP\_ACCEPT 状态，说明是新的客户端接入，则调用 ServerSocketChannel.accept() 方法接受新的客户端；

7、设置新接入的客户端链路 SocketChannel 为非阻塞模式，配置其他的一些 TCP参数；

8、将 SocketChannel 注册到 Selector，监听 OP\_READ 操作位；

9、如果轮询的 Channel为 OP\_READ，则说明 SocketChannel 中有新的就绪的数据包需要读取，则构造 ByteBuffer 对象，读取数据包。

10、如果轮询的 Channel 为 OP\_WRITE ，说明还有数据没有发送完成，需要继续发送。

### NIO服务端序列图 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70 2][]

package com.netty.nio;
    
    public class NIOTimeServer {
        public static void main(String[] args) {
            int port = 8080;
            /**
             * 创建多路复用类，它是一个独立的线程，负责轮询多路复用器 Selector，可以处理多个客户端的并发接入。
             */
            MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
            new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();
        }
    }

package com.netty.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;
    
    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("Server 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.select(1000);
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> it = selectionKeys.iterator();
                    SelectionKey key = null;
                    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();
                }
            }
            if (selector != null) {
                try {
                    selector.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        private void handleInput(SelectionKey key) throws IOException {
            if (key.isValid()) {
                // 处理新接入的请求消息
                if (key.isAcceptable()) {
                    // Accept the new connection
                    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                    SocketChannel sc = channel.accept();
                    sc.configureBlocking(false);
                    // Add the new connection to the selector
                    sc.register(selector, SelectionKey.OP_READ);
                }
                if (key.isReadable()) {
                    // Read data
                    SocketChannel sc = (SocketChannel) key.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    int readBytes = sc.read(buffer);
                    if (readBytes > 0) {
                        buffer.flip();
                        byte[] bytes = new byte[buffer.remaining()];
                        buffer.get(bytes);
                        String body = new String(bytes, "UTF-8");
                        System.out.println("server recv body: " + body);
                        String now = "QUERY TIME ORDER".equals(body) ? new Date(System.currentTimeMillis()).toString()
                                : "BAD ORDER";
                        doWrite(sc, now);
                    } else if (readBytes < 0){
                        // 对端链路关闭
                        key.channel();
                        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);
                writeBuffer.put(bytes);
                writeBuffer.flip();
                channel.write(writeBuffer);
            }
        }
    }

### NIO 客户端创建序列 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70 3][]

package com.netty.nio;
    
    public class NIOTimeClient {
        public static void main(String[] args) {
            int port = 8080;
            new Thread(new TimeClientHandler("127.0.0.1", port), "TimeClient-001").start();
        }
    }

package com.netty.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.SocketChannel;
    import java.util.Iterator;
    import java.util.Set;
    
    public class TimeClientHandler implements Runnable {
        private String host;
        private int port;
        private Selector selector;
        private SocketChannel socketChannel;
        private volatile boolean stop;
    
        public TimeClientHandler(String host, int port) {
            this.host = host;
            this.port = port;
            try {
                selector = Selector.open();
                socketChannel = SocketChannel.open();
                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> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> it = selectionKeys.iterator();
                    SelectionKey key = null;
                    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();
                    System.exit(1);
                }
            }
    
            if (selector != null) {
                try {
                    selector.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        private void handleInput(SelectionKey key) throws IOException {
            if (key.isValid()) {
                SocketChannel sc = (SocketChannel) key.channel();
                if (key.isConnectable()) {
                    if (sc.finishConnect()) {
                        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 {
                        ;
                    }
                }
            }
        }
    
        private void doConnect() throws IOException{
            //  如果连接成功，则注册到多路复用器上，发送请求消息，读应答
            if (socketChannel.connect(new InetSocketAddress(host, port))) {
                socketChannel.register(selector, SelectionKey.OP_READ);
                doWrite(socketChannel);
            } else {
                socketChannel.register(selector, SelectionKey.OP_CONNECT);
            }
        }
    
        private void doWrite(SocketChannel socketChannel) throws IOException {
            byte[] req = "QUERY TIME ORDER".getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
            writeBuffer.put(req);
            writeBuffer.flip();
            socketChannel.write(writeBuffer);
            if (!writeBuffer.hasRemaining()) {
                System.out.println("Send Order to Server succeed.");
            }
        }
    }

使用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。

# AIO #

NIO 2.0 引入了新的异步通道的概念，并提供了异步文件通道和异步套接字通道的实现。异步通道提供一下两种方式获取操作结果。

1、通过 java.util.concurrent.Future 类来表示异步操作的结果；

2、在执行异步操作的时候传入一个 java.nio.channels。

CompletionHandler 接口的实现类作为操作完成的回调。

NIO 2.0 的异步套接字通道是真正的异步非阻塞 I/O，对应于 UNIX网络编程中的事件驱动 I/O （AIO）。它不需要通过多路复用器（Selector）对注册的通道进行轮询操作即可实现异步读写，从而简化了NIO的 编程模型。

服务端代码：

package com.netty.aio;
    
    public class AIOTimeServer {
        public static void main(String[] args) {
            int port = 8080;
    
            new Thread(new AsyncTimeServerHandler(port), "AIO-AsyncTimeServerHandler-001").start();
        }
    }

package com.netty.aio;
    
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.channels.AsynchronousServerSocketChannel;
    import java.util.concurrent.CountDownLatch;
    
    public class AsyncTimeServerHandler implements Runnable {
        private int port;
    
        CountDownLatch countDownLatch;
    
        AsynchronousServerSocketChannel asynchronousServerSocketChannel;
    
        public AsyncTimeServerHandler(int port) {
            this.port = port;
            try {
                asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
                asynchronousServerSocketChannel.bind(new InetSocketAddress(this.port));
                System.out.println("Server start in port : " + this.port);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        @Override
        public void run() {
            countDownLatch = new CountDownLatch(1);
            doAccept();
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        private void doAccept() {
            asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());
        }
    }

package com.netty.aio;
    
    import java.nio.ByteBuffer;
    import java.nio.channels.AsynchronousSocketChannel;
    import java.nio.channels.CompletionHandler;
    
    
    public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler> {
    
        @Override
        public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
            attachment.asynchronousServerSocketChannel.accept(attachment, this);
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            result.read(buffer, buffer, new ReadCompletionHandler(result));
        }
    
        @Override
        public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
            exc.printStackTrace();
            attachment.countDownLatch.countDown();
        }
    }

package com.netty.aio;
    
    import java.io.IOException;
    import java.nio.ByteBuffer;
    import java.nio.channels.AsynchronousSocketChannel;
    import java.nio.channels.CompletionHandler;
    import java.util.Date;
    
    public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {
    
        private AsynchronousSocketChannel channel;
    
        public ReadCompletionHandler(AsynchronousSocketChannel channel) {
            if (this.channel == null)
                this.channel = channel;
        }
    
        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            attachment.flip();
            byte[] body = new byte[attachment.remaining()];
            attachment.get(body);
            try {
                String req = new String(body, "UTF-8");
                System.out.println("Server recv order : " + req);
                String now = "QUERY TIME ORDER".equals(req) ? new Date(System.currentTimeMillis()).toString()
                        : "BAD ORDER";
                doWrite(now);
            } catch (Exception e) {
    
            }
        }
    
        private void doWrite(String now) {
            if (now != null && now.trim().length() > 0) {
                byte[] bytes = now.getBytes();
                ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
                writeBuffer.put(bytes);
                writeBuffer.flip();
                channel.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                    @Override
                    public void completed(Integer result, ByteBuffer attachment) {
                        // 如果没有发送完成，继续发送
                        if (attachment.hasRemaining()) {
                            channel.write(attachment, attachment, this);
                        }
                    }
    
                    @Override
                    public void failed(Throwable exc, ByteBuffer attachment) {
                        try {
                            channel.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        }
    
        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

Client 代码

package com.netty.aio;
    
    public class AIOTimeClient {
        public static void main(String[] args) {
            int port = 8080;
    
            new Thread(new AsyncTimeClientHandler("127.0.0.1", port), "AIO-AsyncTimeClientHandler-001").start();
        }
    }

package com.netty.aio;
    
    
    import java.io.IOException;
    import java.io.UnsupportedEncodingException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.AsynchronousSocketChannel;
    import java.nio.channels.CompletionHandler;
    import java.util.concurrent.CountDownLatch;
    
    public class AsyncTimeClientHandler implements CompletionHandler<Void, AsyncTimeClientHandler>, Runnable {
        private AsynchronousSocketChannel client;
        private String host;
        private int port;
        private CountDownLatch latch;
    
        public AsyncTimeClientHandler(String host, int port) {
            this.host = host;
            this.port = port;
            try {
                client = AsynchronousSocketChannel.open();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        @Override
        public void run() {
            latch = new CountDownLatch(1);
            client.connect(new InetSocketAddress(host, port), this, this);
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        @Override
        public void completed(Void result, AsyncTimeClientHandler attachment) {
            byte[] req = "QUERY TIME ORDER".getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
            writeBuffer.put(req);
            writeBuffer.flip();
            client.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                @Override
                public void completed(Integer result, ByteBuffer attachment) {
                    if (attachment.hasRemaining()) {
                        client.write(attachment, attachment, this);
                    } else {
                        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                        client.read(readBuffer, readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                            @Override
                            public void completed(Integer result, ByteBuffer attachment) {
                                attachment.flip();
                                byte[] bytes = new byte[attachment.remaining()];
                                attachment.get(bytes);
                                String body;
                                try {
                                    body = new String(bytes, "UTF-8");
                                    System.out.println("Now is :" + body);
                                    latch.countDown();
                                } catch (UnsupportedEncodingException e) {
                                    e.printStackTrace();
                                }
                            }
    
                            @Override
                            public void failed(Throwable exc, ByteBuffer attachment) {
                                try {
                                    client.close();
                                    latch.countDown();
                                } catch (IOException e) {
                                    e.printStackTrace();
                                }
                            }
                        });
                    }
                }
    
                @Override
                public void failed(Throwable exc, ByteBuffer attachment) {
                    try {
                        client.close();
                        latch.countDown();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    
        @Override
        public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
            try {
                client.close();
                latch.countDown();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

### 总结： ###

BIO 是同步阻塞读和写，NIO类库支持非阻塞读和写操作。

NIO 底层使用epoll，它是基于I/O多路复用技术的非阻塞I/O，不是异步I/O。多路复用的核心就是通过 Selector 来轮询注册在其上的 Channel，当发现某个或者多个 Channel 处于就绪状态后，从阻塞状态返回就绪的 Channel 的选择键集合，进行I/O 操作。

NIO 2.0 新增了异步的套接字通道，它是真正的异步 I/O，在异步 I/O操作的时候开业传递信号变量，当操作完成之后会回调相关的方法，异步 I/O 也被称为AIO。

# 不同 I/O 模型对比 #

![20190410144244238.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70 4][]

[Link 1]: https://blog.csdn.net/qzcsu/article/details/72861891
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70]: /images/20220222/8b9e8c1db39646e29e3a4826cda80812.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70 1]: /images/20220222/a3950f7a2e894f6d99c2ba46f3ac02e9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70 2]: /images/20220222/cd81a4660b3549daa12ff496798eb7ee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70 3]: /images/20220222/0536a0db50754bdb9b9a0ffc4cc554ee.png
[20190410144244238.png]: /images/20220222/601c3e406cd4445cb98f12135a1080a4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70 4]: /images/20220222/881b4c7d4d1c450e93b46f0aa75f0dfe.png