第二章 NIO入门

  • 传统的同步阻塞式I/O编程
  • 基于NIO的非阻塞编程
  • 基于NIO2.0的异步非阻塞（AIO）编程
  • 为什么要使用NIO编程
  • 为什么选择Netty

　　第二章 NIO 入门

　　　　2.1 传统的BIO编程

　　　　　　2.1.1 BIO 通信模型图

　　　　　　2.1.2 同步阻塞式I/O创建的TimeServer源码分析

package com.phei.netty.bio;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

public class TimeServer {
    public static void main(String[] args)throws IOException{
        int port = 8080;
        if(args != null && args.length > 0){
            try{
                port = Integer.valueOf(port);
            }catch(NumberFormatException e){
//                port = 8080;
            }
        }
        ServerSocket server = null;
        try{
//            如果端口合法且没有被占用，服务端监听成功
            server = new ServerSocket(port);
            System.out.println("The time server is start in port:" + port);
            Socket socket = null;
            while(true){
//                如果没有客户端接入，则主线程阻塞在ServerSocket的accept操作上
                socket = server.accept();
                new Thread(new TimeServerHandler(socket)).start();
            }
        }finally{
            if(server != null){
                System.out.println("The time server close");
                server.close();
                server = null;
            }
        }
    }
}
package com.phei.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() {
        // TODO Auto-generated constructor stub
    }
    
    public TimeServerHandler(Socket socket) {
        super();
        this.socket = socket;
    }

    @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 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;
            }
        }
    }

}

　　　　　　2.1.3 同步阻塞式I/O创建的TimeClient源码分析

package 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 {

    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){
            
        }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;
            }
        }
    }
}

　　　　2.2 伪异步I/O编程

　　　　　　2.2.1 伪异步I/O模型图

　　　　　　2.2.2 伪异步I/O创建的TimeServer源码分析

package com.phei.netty.pio;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

import com.phei.netty.bio.TimeServerHandler;

public class TimeServer {

    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;
            //创建I/O任务线程池
            TimeServerHandlerExecutePool singleExecutor = new TimeServerHandlerExecutePool(50,10000);
            while(true){
                socket = server.accept();
//                当接收到新的客户端连接时，将请求Socket封装成一个Task，然后调用线程池的execute方法执行，从而避免了每个请求接入都创建一个新的线程。
                singleExecutor.execute(new TimeServerHandler(socket));
            }
        }finally{
            if(server != null){
                System.out.println("The time server close");
                server.close();
                server = null;
            }
        }
    }
}
package com.phei.netty.pio;

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);
    }
}

　　　　　　2.2.3 伪异步I/O弊端分析

　　　　2.3 NIO编程

　　　　　　2.3.1 NIO类库简介

　　　　　　　　1.缓冲区Buffer

　　　　　　　　　　ByteBuffer

　　　　　　　　　　CharBuffer

　　　　　　　　　　ShortBuffer

　　　　　　　　　　IntBuffer

　　　　　　　　　　LongBuffer

　　　　　　　　　　FloatBuffer

　　　　　　　　　　DoubleBuffer

　　　　　　　　2.通道Channel

　　　　　　　　　　网络读写：SelectableChannel

　　　　　　　　　　文件操作：FileChannel

　　　　　　　　3.多路复用器Selector

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

　　　　　　2.3.2 NIO服务端序列图

　　　　　　2.3.3 NIO创建的TimeServer源码分析

package 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。它是一个独立的线程，负责轮询多路复用器Selctor，可以处理多个客户端的并发接入。
        MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
        new Thread(timeServer,"NIO-MultiplexerTimeServer-001").start();
    }
}
package 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.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;
    
    /*
     * 初始化多路复用器，绑定监听端口
     * 在构造方法中创建多路复用器Selector、ServerSocketChannel，对Channel和TCP参数进行配置。
     * 将ServlerSocketChannel设置为异步非阻塞模式，它的backlog设为1024。
     * 系统资源初始化成功后，将ServerSocketChannel注册到Selector，监听SelectionKey.OP_ACCEPT操作位。
     * 如果资源初始化失败，则退出。
     */
    public MultiplexerTimeServer(int port){
        try{
            //Opens a selector.
            selector = Selector.open();
            //Opens a server-socket channel.
            servChannel = ServerSocketChannel.open();
//            Adjusts this channel's non-blocking mode.
            servChannel.configureBlocking(false);
//            Retrieves a server socket associated with this channel.
//            Binds the ServerSocket to a specific address (IP address and port number).
//            1024 ： requested maximum length of the queue of incoming connections.
            servChannel.socket().bind(new InetSocketAddress(port),1024);
//            Registers this channel with the given selector, returning a selection key.
//            SelectionKey.OP_ACCEPT : Operation-set bit for socket-accept operations.
            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;
    }
    
    /**
     * 在线程的run方法的while循环体中循环遍历selector，它的休眠时间为1s。
     * 无论是否有读写等事件发生，selector每隔1s都被唤醒一次。
     * selector也提供了一个无参的select方法：当有处于就绪状态的Channel时，selector将返回该Channel的SelectionKey集合。
     * 通过对就绪状态的Channel集合进行迭代，可以进行网络的异步读写操作。
     */
    @Override
    public void run() {
        while(!stop){
            try{
//                Selects a set of keys whose corresponding channels are ready for I/O operations.
//                timeout - If positive, block for up to timeout milliseconds, more or less, while waiting for a channel to become ready; 
// 　　　　　　　　 if zero, block indefinitely; must not be negative
                selector.select(1000);
//                Returns this selector's selected-key set.
                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){
//                            Requests that the registration of this key's channel with its selector be cancelled.
                            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();
            }
        }
    }
    /*
     * 处理新接入的客户端请求信息，根据SelectionKey的操作位进行判断即可获知网络时间的类型，
     * 通过ServerSocketChannel的accept接收客户端的连接请求并创建SocketChannel实例。
     * 完成之后相当于完成了TCP的三次握手，TCP物理链路正是建立。
     * 需要将新创建的SocketChannel设置为异步非阻塞，同时也可以对其TCP参数进行设置，如TCP接收和发送缓冲区的大小等。
     */
    private void handleInput(SelectionKey key) throws IOException{
//        Tells whether or not this key is valid.
        if(key.isValid()){
//            Tests whether this key's channel is ready to accept a new socket connection.
//            处理新接入的请求消息
            if(key.isAcceptable()){
//                Returns the channel for which this key was created.
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
//                Accepts a connection made to this channel's socket.
                SocketChannel sc = ssc.accept();
//                Adjusts this channel's blocking mode.
                sc.configureBlocking(false);
//                Registers this channel with the given selector, returning a selection key.
//                The interest set for the resulting key
                sc.register(selector, SelectionKey.OP_READ);
            }
            /*
             * 用于读取客户端的请求消息。
             * 首先创建一个ByteBuffer，由于事先无法得知客户端发送的码流大小，作为历程，开辟一个1MB的缓冲区。
             * 然后调用SocketChannel的read方法读取请求码流。
             */
//            Tests whether this key's channel is ready for reading.
            if(key.isReadable()){
                SocketChannel sc = (SocketChannel) key.channel();
//                Allocates a new byte buffer.
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
//                Reads a sequence of bytes from this channel into the given buffer.
                int readBytes =sc.read(readBuffer);
                if(readBytes > 0){
//                    对readBuffer进行flip操作，将缓冲区当前的limit设置为position，position设置为0，用于后续对缓冲区的读取操作。
//                    Flips this buffer.
                    readBuffer.flip();
//                    根据缓冲区刻度的字节个数创建字节数组
//                    Returns the number of elements between the current position and the limit.
                    byte[] bytes = new byte[readBuffer.remaining()];
//                    调用ByteBuffer的get操作将缓冲区可读字节数组复制到新创建的字节数组中
//                    Relative bulk get method.
                    readBuffer.get(bytes);
                    String body = new String(bytes,"UTF-8");
                    System.out.println("The time server receive order : " + body);
                    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字节，忽略
                    ;
                }
            }
        }
    }
    /*
     * 将应答消息异步发送给客户端。
     * 由于SocketChannel是异步非阻塞的，它并不保证一次能够把需要发送的字节数组发送完，此时会出现“写半包”问题。
     * 需要注册写操作，不断轮询Selector将没有发送完的ByteBuffer发送完毕，
     * 然后可以通过ByteBuffer的hasRemain()方法判断消息是否发送完成。
     */
    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);
//            Relative bulk put method  (optional operation).
            writeBuffer.put(bytes);
            writeBuffer.flip();
//            调用SocketChannel的write方法将缓冲区中的字节数组发送出去
//            Writes a sequence of bytes to this channel from the given buffer.
            channel.write(writeBuffer);
        }
    }
}

　　　　　　2.3.4 NIO客户端序列图

　　　　　　2.3.5 NIO创建的TimeClient源码分析

package 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 TimeClientHandle("127.0.0.1",port),"TimeClient-001").start();
    }
}
package 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.util.Iterator;
import java.util.Set;

public class TimeClientHandle implements Runnable {

    private String host;
    private int port;
    private Selector selector;
    private SocketChannel socketChannel;
    private volatile boolean stop;
    
    public TimeClientHandle() {
        super();
        // TODO Auto-generated constructor stub
    }

    /*
     * 使用构造函数初始化NIO的多路复用器和SocketChannel对象。
     * 创建SocketChannel之后将其设置为异步非阻塞模式。
     * 在此可以设置SocketChannel的TCP参数
     */
    public TimeClientHandle(String string, 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);
        }
    }

    /*
     * 作为示例，连接是成功的，所以不需要做重连操作，因此将其放到循环之前。
     */
    @Override
    public void run() {
        try{
//            如果连接成功、如果没有成功
            doConnect();
        }catch(IOException e){
            e.printStackTrace();
            System.exit(1);
        }
//        轮询多路复用器Selector。当有就绪的Channel时，执行handleInput(key)方法。
        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{
//                        当有就绪的Channel时执行
                        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{
//        Tells whether or not this key is valid.
        if(key.isValid()){
//            Returns the channel for which this key was created.
            SocketChannel sc = (SocketChannel) key.channel();
//            判断是否连接成功
//            如果处于连接状态，说明服务端已经返回ACK应答消息。
//            Tests whether this key's channel has either finished, or failed to finish, 
//            its socket-connection operation.
            if(key.isConnectable()){
//                调用SecoketChannel的finishConnect()方法。
//                如果返回值为true，说明客户端连接成功；
//                如果返回值为false或者抛出IOException，说明连接失败
//                Finishes the process of connecting a socket channel.
                if(sc.finishConnect()){
//                    将SocketChannel注册到多路复用器上，注册SelectionKey.OP_READ操作位，监听网络读操作，
//                    然后发送请求消息给服务端
                    sc.register(selector, SelectionKey.OP_READ);
//                    构造请求消息体，然后对其编码，写入到发送缓冲区中，最后调用SocketChannel的write方法进行发送。存在"半包写"
//                    最后通过hasRemaining()方法对发送结果进行判断
                    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{
//        如果直接连接成功，则注册到多路复用器上，发送请求消息，读应答
//        连接此通道的套接字
        if(socketChannel.connect(new InetSocketAddress(host, port))){
//            连接成功，将SocketChannel注册到多路复用器Selector上，注册SelectionKey.OP_READ
            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.");
        }
    }
}

　　　　　服务端控制台：

The time server is start in port:8080
The time server receive order : QUERY TIME ORDER

　　　　　客户端控制台：

Send order 2 server succeed.

　　　　　　socketChannel.connect(new InetSocketAddress(host, port)：返回false

　　　　　　key.isReadable()：返回false

　　　　　　WHY？？？？？？？？？？？？？？？？？？？？？？？？？？？？

　　　　　2.4 AIO编程

　　　　　　NIO 2.0 引入了新的异步通道的概念，并提供了异步文件通道和异步套接字通道的实现。

　　　　　　异步通道提供一下两种方式获取操作结果：

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

　　　　　　　　在执行异步操作的时候传入一个 java.nio.channels.CompletionHandler接口的实现类作为操作完成的回调。

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

　　　　　　2.4.1 AIO创建的TimeServer源码分析

package com.phei.netty.aio;

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){
                //采用默认值
            }
        }
//        创建异步的时间服务器处理类
        AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port);
//        启动线程
        new Thread(timeServer,"AIO-AsyncTimeServerHandler-001").start();
    }
}
package com.phei.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 latch;
    AsynchronousServerSocketChannel asynchronousServerSocketChannel;
    
//    创建一个异步的服务端通道AsynchronousServerSocketChannel，然后调用它的bind方法绑定监听端口。
    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();
        }
    }
    @Override
    public void run() {
//        在完成一组正在执行的操作之前，允许当前的线程一直阻塞。
//        在本例中，我们让线程在此阻塞，防止服务端执行完成退出。
//        在实际项目应用中，不需要启动独立的线程来处理AsynchronousServerSocketChannel，这里仅仅是个demo演示
        latch = new CountDownLatch(1);
//        用于接收客户端的连接，由于是异步操作，可以传递一个CompletionHandler<SsynchronousSocketChannel,? super A>类型的handler实例接收accept操作成功的通知消息
        doAccept();
        try{
            latch.await();
        }catch(InterruptedException e){
            e.printStackTrace();
        }
    }
    private void doAccept() {
        asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());
    }

}
package com.phei.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方法
//        调用AsynchronousServerSocketChannel的accept方法后，
//        如果有新的客户端连接接入，系统将回调传入的CompletionHandler实例的completed方法，表示新的客户端已经接入成功。
//        因为一个AsynchronousServerSocketChannel可以接收成千上万个客户端，所以需要继续调用它的accept方法，
//        接收其他的客户端连接，最终形成一个循环。每当接收一个客户读连接成功之后，在异步连接新的客户端连接。
        attachment.asynchronousServerSocketChannel.accept(attachment, this);
//        链路建立成功之后，服务端需要接收客户端的请求消息。
//        创建新的ByteBuffer，预分配1MB的缓冲区。
        ByteBuffer buffer = ByteBuffer.allocate(1024);
//        通过调用AsynchronousSocketChannel的read方法进行异步读操作。
//        ByteBuffer dst : 接收缓冲区，用于从异步Channel中读取数据包；
//        A attachment : 异步Channel携带的附件，通知回调的时候作为入参使用；
//        CompletionHandler<Integer,? super A> : 接收通知回调的业务Handler，在例程中为ReadCOmpletionHandler
        result.read(buffer,buffer,new ReadCompletionHandler(result));
    }

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

}
package 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;
import java.util.Date;

public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {

    private AsynchronousSocketChannel channel;
    
//    将AsynchronousSocketChannel通过参数传递到ReadCompletionHandler中，
//    当作成员变量来使用，主要用于读取半包消息和发送应答
    public ReadCompletionHandler(AsynchronousSocketChannel channel) {
        if(this.channel == null){
            this.channel = channel;
        }
    }

//    读取到消息后的处理
    @Override
    public void completed(Integer result, ByteBuffer attachment) {
//        对attachment进行flip操作，为后续冲缓冲区读取数据做准备。
        attachment.flip();
//        根据缓冲区的刻度字节数创建byte数组
        byte[] body = new byte[attachment.remaining()];
        attachment.get(body);
        try{
//            通过new String方法创建请求消息，对请求消息进行判断
            String req = new String(body,"UTF-8");
            System.out.print("The time server receive order : " + req);
            String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req) ? new Date(System.currentTimeMillis()).toString() : "BAD ORDER";
//            调用doWrite方法发送给客户端
            doWriter(currentTime);
        }catch(UnsupportedEncodingException e){
            e.printStackTrace();
        }
    }

    private void doWriter(String currentTime) {
//        对当前事件进行合法性校验
        if(currentTime != null && currentTime.trim().length() > 0){
//            调用字符串的解码方法将应答消息编码成字节数组，然后将它复制到发送缓冲区writeBuffer中
            byte[] bytes = currentTime.getBytes();
            ByteBuffer writerBuffer = ByteBuffer.allocate(bytes.length);
            writerBuffer.put(bytes);
            writerBuffer.flip();
//            调用AsynchronousSocketChannel的异步write方法
            channel.write(writerBuffer,writerBuffer,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){
                        //ingonre on close
                    }
                }
            });
        }
    }

//    当发送异常的时候，对异常Throwable进行判断：如果I/O异常，就关闭链路，释放资源；
//    如果是其他异常，按照业务自己的逻辑进行处理。本例程作为简单的demo，没有对异常进行分类判断，只要发生了读写异常，就关闭链路，释放资源。
    @Override
    public void failed(Throwable exc, ByteBuffer attachment) {
        try{
            this.channel.close();
        }catch(IOException e){
            e.printStackTrace();
        }
    }

}

　　　　　　2.4.2 AIO创建的TimeClient源码分析

　　　　　　　　Class ：TimeClient

package com.phei.netty.aio;


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){
                
            }
        }
//        通过一个独立的I/O线程常见异步时间服务器客户端Handler。
//        在实际项目中，我们不需要独立的线程创建异步连接对象，因为底层都是通过JDK的系统回调实现的，
//        在后面运行时间服务器程序的时候，我们会抓取线程调用堆栈给大家展示
        new Thread(new AsyncTimeClientHandler("127.0.0.1",port),"AIO-AsyncTimeClientHandler-001").start();
    }
}

　　　　　　　　Class ： AsyncTimeClientHandler

package 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;

public class AsyncTimeClientHandler implements
        CompletionHandler<Void, AsyncTimeClientHandler> , Runnable {

    private AsynchronousSocketChannel client;
    private String host;
    private int port;
    private CountDownLatch latch;
    
    public AsyncTimeClientHandler() {
        super();
    }

    public AsyncTimeClientHandler(String host, int port) {
        super();
        this.host = host;
        this.port = port;
        try{
//    1.通过AsynchronousSocketChannel的open方法创建一个新的AsynchronousSocketChannel对象。
            client = AsynchronousSocketChannel.open();
        }catch(IOException e){
            e.printStackTrace();
        }
    }

    public void run() {
//        2.创建CountDownLatch进行等待，防止异步操作没有执行完成线程就退出
        latch = new CountDownLatch(1);
//        3.通过connect方法发起异步操作
//        A attachment ： AsynchronousSocketChannel 的附件，用于回调通知时作为入参被传递，调用者可自定义。
//        CompletionHandler<Void,? super A> handler ： 异步操作回调通知接口，由调用者实现
//        在本例程中，这两个参数都使用AsyncTimeClientHandler类本身，因为它实现了CompletionHandler接口
        client.connect(new InetSocketAddress(host,port),this,this);
        try{
            latch.await();
        }catch(InterruptedException e1){
            e1.printStackTrace();
        }
        try{
            client.close();
        }catch(IOException e){
            e.printStackTrace();
        }
    }

//    4.异步连接成功之后的方法回调completed。
    public void completed(Void result, AsyncTimeClientHandler attachment) {
//        创建请求消息体，对其进行编码。
        byte[] req ="QUERY TIME ORDER".getBytes();
        ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
//        复制到发送缓冲区writeBuffer中，
        writeBuffer.put(req);
        writeBuffer.flip();
//        调用AsynchronousSocketChannel的write方法进行异步写。
//        实现CompletionHandler<Integer,ByteBuffer>接口用于写操作完成后的回调
        client.write(writeBuffer,writeBuffer,new CompletionHandler<Integer,ByteBuffer>(){

            public void completed(Integer result, ByteBuffer buffer) {
                if(buffer.hasRemaining()){
                    client.write(buffer,buffer,this);
                }else{
                    ByteBuffer readBuffer = ByteBuffer.allocate(1024);
//                    5.客户端异步读取时间服务器服务端应答消息的处理逻辑。
//                    调用AsynchronousSocketChannel的read方法异步读取服务端的响应消息
//                    由于read操作是异步的，所以我们通过内部匿名类实现CompletionHandler<Integer,ByteBuffer>接口，
//                    当读取完成被JDK回调时，构造应答消息。
                    client.read(readBuffer,readBuffer,new CompletionHandler<Integer,ByteBuffer>(){

                        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();
                            }
                        }

                        public void failed(Throwable exc, ByteBuffer attachment) {
                            try{
                                client.close();
                                latch.countDown();
                            }catch(IOException e){
                                // ingnore on close
                            }
                        }
                    });
                }
            }

//            当读取发生异常时，关闭链路，同时调用CountDownLatch的countDown方法让AsyncTimeClientHandler线程执行完毕，客户端退出执行
            public void failed(Throwable exc, ByteBuffer attachment) {
                try{
                    client.close();
                    latch.countDown();
                }catch(IOException e){
                    //ingnore on close
                }
            }
        });
    }

    public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
        exc.printStackTrace();
        try{
            client.close();
            latch.countDown();
        }catch(IOException e){
            e.printStackTrace();
        }
    }
}

　　　　　　2.4.3 AIO版本时服务器的运行结果

　　　　2.5 4中I/O的对比

　　　　　　2.5.1 概念澄清

　　　　　　　　1.异步非阻塞I/O

　　　　　　　　2.多路复用器Selector

　　　　　　　　3.伪异步I/O

　　　　　　2.5.2 不同I/O模型对比

  同步阻塞I/O(BIO) 伪异步I/O 非阻塞I/O(NIO) 异步I/O(AIO)
客户端个数：I/O线程 1：1 M：N(其中M可以大于N) M:1(1个I/O线程处理多个客户端连接) M：0(不需要启动额外的I/O线程，被动回调)
I/O类型(阻塞) 阻塞I/O 阻塞I/O 非阻塞I/O 非阻塞I/O
I/O类型(同步) 同步I/O 同步I/O 同步I/O(I/O多路复用) 异步I/O
API使用难度 简单 简单 非常复杂 复杂
调试难度 简单 简单 高 高
可靠性 非常差 差 高 高
吞吐量 低 中 高 高

　　　　2.6 选择Netty的理由

　　　　　　2.6.1 不选择Java原生NIO编程的原因

　　　　　　2.6.2 为什么选择Netty

　　　　2.7 总结

　　　　

　　

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/ClassNotFoundException/p/6212107.html