Netty之NIO入门（二）

小咪咪 2023-03-01 05:51 202阅读 0赞

### 文章目录 ###

*   *  BIO
     *   *  \`思考\`
     *  NIO
     *   *   *  分析
     *  NIO2.0 （AIO）
     *  最后

## BIO ##

下图的通信模型图来熟悉下BIO的服务端通信模型：采用BIO通信模型的服务端，通常由一个独立的 Acceptor线程负责监听客户端的连接，它接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理，处理完成之后，通过输出流返回应答给客户端，线程销毁。这就是典型的一请求一应答通信模型。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70]  
该模型最大的问题就是缺乏弹性伸缩能力，当客户端并发访问量增加后，服务端的线程个数和客户端并发访问数呈1:1的正比关系，由于线程是Java虚拟机非常宝贵的系统资源，当线程数膨胀之后，系统的性能将急剧下降，随着并发访问量的继续增大，系统会发生线程堆栈溢出、创建新线程失败等问题，并最终导致进程宕机或者僵死，不能对外提供服务。

**同步阻塞式/O创建的 Time Server源码分析**

package main.com.phei.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 { 
    
        /** * @param args */
        public static void main(String[] args) { 
    
            int port = 8080;
            if (args != null && args.length > 0) { 
    
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
    
            }
            Socket socket = null;
            BufferedReader in = null;
            PrintWriter out = null;
            try { 
                socket = new Socket("127.0.0.1", port);
                in = new BufferedReader(new InputStreamReader(
                        socket.getInputStream()));
                out = new PrintWriter(socket.getOutputStream(), true);
                out.println("QUERY TIME ORDER");
                System.out.println("Send order 2 server succeed.");
                String resp = in.readLine();
                System.out.println("Now is : " + resp);
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                if (out != null) { 
                    out.close();
                    out = null;
                }
    
                if (in != null) { 
                    try { 
                        in.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                    in = null;
                }
    
                if (socket != null) { 
                    try { 
                        socket.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                    socket = null;
                }
            }
        }
    }

`以上是服务端，默认设置监听端口8080，然后通过一个死循环监听客户端连接，如果没有客户端连接就会阻塞在accept上。 当有新的客户端接入的时候，执行代码34行，以 Socket为参数构造 Timeserverhandler对象， Time Serverhandler是一个 Runnable，使用它为构造函数的参数创建一个新的客户端线程处理这条 Socket链路。下面我们继续分析 Timeserverhandler的代码。`

package main.com.phei.netty.bio;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.io.PrintWriter;
    import java.net.Socket;
    
    public class TimeServerHandler implements Runnable { 
    
        private Socket socket;
    
       public TimeServerHandler(Socket socket) { 
            this.socket = socket;
        }
    
        /* * (non-Javadoc) * * @see java.lang.Runnable#run() */
        @Override
        public void run() { 
            BufferedReader in = null;
            PrintWriter out = null;
            try { 
                in = new BufferedReader(new InputStreamReader(
                        this.socket.getInputStream()));
                out = new PrintWriter(this.socket.getOutputStream(), true);
                String currentTime = null;
                String body = null;
                while (true) { 
                    body = in.readLine();
                    if (body == null) { 
                        break;
                    }
                    System.out.println("The time server receive order : " + body);
                    currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(
                            System.currentTimeMillis()).toString() : "BAD ORDER";
                    out.println(currentTime);
                }
    
            } catch (Exception e) { 
                if (in != null) { 
                    try { 
                        in.close();
                    } catch (IOException e1) { 
                        e1.printStackTrace();
                    }
                }
                if (out != null) { 
                    out.close();
                    out = null;
                }
                if (this.socket != null) { 
                    try { 
                        this.socket.close();
                    } catch (IOException e1) { 
                        e1.printStackTrace();
                    }
                    this.socket = null;
                }
            }
        }
    }

`这一段拿到socket的输入流然后读取内容，如果是QUERY TIME ORDER就把时间写入到一个输出流返回给客户端。`

package main.com.phei.netty.bio;
    
    import java.io.IOException;
    import java.net.ServerSocket;
    import java.net.Socket;
    
    public class TimeServer { 
    
        /** * @param args * @throws IOException */
        public static void main(String[] args) throws IOException { 
            int port = 8080;
            if (args != null && args.length > 0) { 
    
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
    
            }
            ServerSocket server = null;
            try { 
                server = new ServerSocket(port);
                System.out.println("The time server is start in port : " + port);
                Socket socket = null;
                while (true) { 
                    socket = server.accept();
                    new Thread(new TimeServerHandler(socket)).start();
                }
            } finally { 
                if (server != null) { 
                    System.out.println("The time server close");
                    server.close();
                    server = null;
                }
            }
        }
    }

运行结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 1]

![在这里插入图片描述][20200722151656860.png]  
**总结**  
BIO主要的问题在于每当有一个新的容户端请求接入时，服务端必须创建一个新的线程处理新接入的客户端链路，一个线程只能处理一个客户端连接。在高性能服务器应用领域，往往需要面向成千上万个客户端的并发连接，这种模型显然无法满足高性能、高并发接入的场景。

### `思考` ###

我们启动一个服务端，然后阻塞在那里，这个时候来了很多连接，然后服务端起了很多线程处理，这看上去没什么问题，而再仔细思考下，我们的主线程其实大部分时间是空闲的（accept阻塞在那），如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 2]  
那么问题来了，我们能不能把另开线程里面执行的东西放到主线程里面执行，因为线程资源是很稀缺的，也就是多个线程里面的任务在主线程里面执行，也就是主线程被重复使用了，也就是`多路复用`,也就出现了我们的NIO

## NIO ##

上代码：

package main.com.phei.netty.nio;
    
    public class TimeClient { 
    
        public static void main(String[] args) { 
    
            int port = 8080;
            if (args != null && args.length > 0) { 
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
            }
            new Thread(new TimeClientHandler("127.0.0.1", port), "TimeClient-001")
                    .start();
        }
    }

package main.com.phei.netty.nio;
    
    public class TimeServer { 
    
        public static void main(String[] args) { 
            int port = 8080;
            if (args != null && args.length > 0) { 
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
            }
            MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
            new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();
        }
    }

package main.com.phei.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.nio.charset.StandardCharsets;
    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 == null ? "127.0.0.1" : host;
            this.port = port;
            try { 
                selector = Selector.open();
                socketChannel = SocketChannel.open();
                socketChannel.configureBlocking(false);
            } catch (IOException e) { 
                e.printStackTrace();
                System.exit(1);
            }
        }
    
        /* * (non-Javadoc) * * @see java.lang.Runnable#run() */
        @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()) { 
                        key = it.next();
                        it.remove();
                        try { 
                            handleInput(key);
                        } catch (Exception e) { 
                            if (key != null) { 
                                key.cancel();
                                if (key.channel() != null)
                                    key.channel().close();
                            }
                        }
                    }
                } catch (Exception e) { 
                    e.printStackTrace();
                    System.exit(1);
                }
            }
    
            // 多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源
            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, StandardCharsets.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 { 
            // 如果直接连接成功，则注册到多路复用器上，发送请求消息，读应答
            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 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.");
        }
    
    }

package main.com.phei.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.nio.charset.StandardCharsets;
    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("The time server is start in port : " + port);
            } catch (IOException e) { 
                e.printStackTrace();
                System.exit(1);
            }
        }
    
        public void stop() { 
            this.stop = true;
        }
    
        /* * (non-Javadoc) * * @see java.lang.Runnable#run() */
        @Override
        public void run() { 
            while (!stop) { 
                try { 
                    selector.select(1000);
                    Set<SelectionKey> selectedKeys = selector.selectedKeys();
                    Iterator<SelectionKey> it = selectedKeys.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 (Throwable t) { 
                    t.printStackTrace();
                }
            }
    
            // 多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源
            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 ssc = (ServerSocketChannel) key.channel();
                    SocketChannel sc = ssc.accept();
                    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();
                    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, StandardCharsets.UTF_8);
                        System.out.println("The time server receive order : "
                                + body);
                        String currentTime = "QUERY TIME ORDER"
                                .equalsIgnoreCase(body) ? new java.util.Date(
                                System.currentTimeMillis()).toString()
                                : "BAD ORDER";
                        doWrite(sc, currentTime);
                    } else if (readBytes < 0) { 
                        // 对端链路关闭
                        key.cancel();
                        sc.close();
                    }
                    // 读到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);
            }
        }
    }

先启动服务端，再启动客户端，运行结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 4]

#### 分析 ####

代码执行完，一脸蒙蔽：  
大概可以用下面的流程来理解：  
有一个selector 这上面可以注册四个事件：OP\_READ ，OP\_WRITE ，OP\_CONNECT， OP\_ACCEPT，并且他本身也在遍历自己的事件，如果有事件注册进来，就进行相应的操作，而他自己先注册一个OP\_ACCEPT事件到自身，用来等待连接，这个时候客户端发来一个连接，并且往通道里面输入一个string值，然后服务端遍历到这个OP\_CONNECT然后往自身注册一个OP\_READ ，然后读取数据。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 5]  
所以现在我们再来看网上的一些文章里面讲的NIO的组件，  
**Selector**  
一个死循环用来读取注册到其中的key，如果有就进行相应的操作  
**Channel**  
负责连接，读取，写入，接受连接的通道，write,read方法用来读写，  
**ByteBuf**  
存在channel中，替代之前BIO中的字节数组

**最后**  
确如在BIO最后思考的那样，最有操作都在一个线程里面执行了，达到了多路复用。

> 几乎所有的中文技术书籍都将 Selector翻译为选择器，但是实际上我认为这样的翻译并不恰当，选择器仅仅是字面上的意思，体现不出 Selector的功能和特点。在前面的章节我们介绍过 Java NIO的实现关键是多路复用I/O技术，多路复用的核心就是通过 Selector来轮询注册在其上的 Channel，当发现某个或者多个 Channel处于就绪状态后，从阻塞状态返回就绪的 Channel的选择键集合，进行I/O操作。由于多路复用器是NIO实现非阻塞1/O的关键，它又是主要通过 Selector实现的，所以本书将 Selector翻译为多路复用器，与其他技术书籍所说的选择器是同一个东西，请大家了解。 -----Netty权威指南

## NIO2.0 （AIO） ##

package main.com.phei.netty.aio;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class TimeClient { 
    
        /** * @param args */
        public static void main(String[] args) { 
            int port = 8080;
            if (args != null && args.length > 0) { 
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
    
            }
            new Thread(new AsyncTimeClientHandler("127.0.0.1", port),
                    "AIO-AsyncTimeClientHandler-001").start();
    
        }
    }

package main.com.phei.netty.aio;
    
    import java.io.IOException;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class TimeServer { 
    
        /** * @param args * @throws IOException */
        public static void main(String[] args) throws IOException { 
            int port = 8080;
            if (args != null && args.length > 0) { 
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
            }
            AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port);
            new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start();
        }
    }

package main.com.phei.netty.aio;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.channels.AsynchronousServerSocketChannel;
    import java.util.concurrent.CountDownLatch;
    
    /** * @author Administrator * @version 1.0 * @date 2014年2月16日 */
    public class AsyncTimeServerHandler implements Runnable { 
    
        CountDownLatch latch;
        AsynchronousServerSocketChannel asynchronousServerSocketChannel;
        private int port;
    
        public AsyncTimeServerHandler(int port) { 
            this.port = port;
            try { 
                asynchronousServerSocketChannel = AsynchronousServerSocketChannel
                        .open();
                asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
                System.out.println("The time server is start in port : " + port);
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    
        /* * (non-Javadoc) * * @see java.lang.Runnable#run() */
        @Override
        public void run() { 
    
            latch = new CountDownLatch(1);
            doAccept();
            try { 
                latch.await();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        public void doAccept() { 
            asynchronousServerSocketChannel.accept(this,
                    new AcceptCompletionHandler());
        }
    
    }

package main.com.phei.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;
    
    /** * @author Administrator * @version 1.0 * @date 2014年2月16日 */
    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 e1) { 
                e1.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 buffer) { 
                            if (buffer.hasRemaining()) { 
                                client.write(buffer, buffer, this);
                            } else { 
                                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                                client.read(
                                        readBuffer,
                                        readBuffer,
                                        new CompletionHandler<Integer, ByteBuffer>() { 
                                            @Override
                                            public void completed(Integer result,
                                                                  ByteBuffer buffer) { 
                                                buffer.flip();
                                                byte[] bytes = new byte[buffer
                                                        .remaining()];
                                                buffer.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) { 
                                                    // ingnore on close
                                                }
                                            }
                                        });
                            }
                        }
    
                        @Override
                        public void failed(Throwable exc, ByteBuffer attachment) { 
                            try { 
                                client.close();
                                latch.countDown();
                            } catch (IOException e) { 
                                // ingnore on close
                            }
                        }
                    });
        }
    
        @Override
        public void failed(Throwable exc, AsyncTimeClientHandler attachment) { 
            exc.printStackTrace();
            try { 
                client.close();
                latch.countDown();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    
    }

package main.com.phei.netty.aio;
    
    import java.nio.ByteBuffer;
    import java.nio.channels.AsynchronousSocketChannel;
    import java.nio.channels.CompletionHandler;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月16日 */
    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.latch.countDown();
        }
    
    }

package main.com.phei.netty.aio;
    
    import java.io.IOException;
    import java.io.UnsupportedEncodingException;
    import java.nio.ByteBuffer;
    import java.nio.channels.AsynchronousSocketChannel;
    import java.nio.channels.CompletionHandler;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月16日 */
    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("The time server receive order : " + req);
                String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req) ? new java.util.Date(
                        System.currentTimeMillis()).toString() : "BAD ORDER";
                doWrite(currentTime);
            } catch (UnsupportedEncodingException e) { 
                e.printStackTrace();
            }
        }
    
        private void doWrite(String currentTime) { 
            if (currentTime != null && currentTime.trim().length() > 0) { 
                byte[] bytes = (currentTime).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 buffer) { 
                                // 如果没有发送完成，继续发送
                                if (buffer.hasRemaining())
                                    channel.write(buffer, buffer, this);
                            }
    
                            @Override
                            public void failed(Throwable exc, ByteBuffer attachment) { 
                                try { 
                                    channel.close();
                                } catch (IOException e) { 
                                    // ingnore on close
                                }
                            }
                        });
            }
        }
    
        @Override
        public void failed(Throwable exc, ByteBuffer attachment) { 
            try { 
                this.channel.close();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    
    }

运行结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 7]

## 最后 ##

下面是几种I/O模型的功能和特性对比：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 8]

> [Netty之Java I/O演进之路（一）][Netty_Java I_O]  
> [Netty之NIO入门（二）][Netty_NIO]  
> [Netty之Netty入门应用（三）][Netty_Netty]  
> [Netty之TCP粘包/拆包问题（四）][Netty_TCP]  
> [Netty之编解码开发（五）][Netty]  
> [Netty之ChannelPipeline(六)][Netty_ChannelPipeline]  
> [Netty之EventLoop和EventLoopGroup(七)][Netty_EventLoop_EventLoopGroup]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70]: /images/20230209/f4419f52cd124f508f1d5535ed3434c7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 1]: /images/20230209/b9a533455539464e854337651e17c4d1.png
[20200722151656860.png]: /images/20230209/154831dd9d77482585b81666d6cb9027.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 2]: /images/20230209/9fe15cf40234492dafc9e4cc58f6ecf7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 3]: /images/20230209/f0e94ff1027c4a9fb80fdd95cf92f325.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 4]: /images/20230209/9ee0342bc7e445eeaa20e1313fff00bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 5]: /images/20230209/9701025f8ff44301b525e1d74ded091b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 6]: /images/20230209/f84c504058644ade8c0f0ea5a5716f60.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 7]: /images/20230209/9df98e69514345e2b073455ce6344cb6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 8]: /images/20230209/38b7ab3f0dd1488a99ae678b1908791c.png
[Netty_Java I_O]: https://blog.csdn.net/jianjun_fei/article/details/107505729
[Netty_NIO]: https://blog.csdn.net/jianjun_fei/article/details/107513219
[Netty_Netty]: https://blog.csdn.net/jianjun_fei/article/details/107555647
[Netty_TCP]: https://blog.csdn.net/jianjun_fei/article/details/107563110
[Netty]: https://blog.csdn.net/jianjun_fei/article/details/107772642
[Netty_ChannelPipeline]: https://blog.csdn.net/jianjun_fei/article/details/107787864
[Netty_EventLoop_EventLoopGroup]: https://blog.csdn.net/jianjun_fei/article/details/107956949