Java NIO学习笔记(四) 使用JDK 1.7 NIO2.0 实现客户端与服务器的通信

梦里梦外; 2022-05-24 03:38 131阅读 0赞

JDK1.7 提供了全新的异步NIO模式。称为：NIO2.0或AIO。该模式引入了新的异步通道的概念，并提供了异步文件通道和异步套接字通道的实现。异步通道提供两种方式获取获取操作结果。分别是：

>  *  通过java.util.concurrent.Future类来表示异步操作的结果；
>  *  CompletionHandler接口的实现类作为操作完成的回调。

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

### 回调模式 ###

#### AIO的回调模式的服务端代码： ####

public class SocketServiceCb {
        
    
        AsynchronousServerSocketChannel asynchronousServerSocketChannel;
        CountDownLatch latch;
    
        public void start() {
    
            try {
    
                asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
                asynchronousServerSocketChannel.bind(new InetSocketAddress("127.0.0.1", 17777));
            } catch (IOException e) {
                e.printStackTrace();
            }
    
            latch = new CountDownLatch(1);
    
            doAccept();
    
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
        }
    
        private void doAccept() {
            asynchronousServerSocketChannel.accept(this, new AccessCompleteHandler());
        }
    
    
        class AccessCompleteHandler implements CompletionHandler<AsynchronousSocketChannel, SocketServiceCb> {
    
            @Override
            public void completed(AsynchronousSocketChannel result, SocketServiceCb attachment) {
    
                // 当我们调用AsynchronousServerSocketChannel的accept方法后，如果有新的客户端连接接入，
                // 系统将回调我们传入的CompletionHandler实例的completed方法，表示新的客户端已经接入成功，
                // 因为一个AsynchronousServerSocket Channel可以接收成千上万个客户端，
                // 所以我们需要继续调用它的accept方法，接收其他的客户端连接，
                // 最终形成一个循环。每当接收一个客户读连接成功之后，再异步接收新的客户端连接。
                attachment.asynchronousServerSocketChannel.accept(attachment, this);
    
                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    
                byteBuffer.flip();
                //参数1 ByteBuffer dst：接收缓冲区，用于从异步Channel中读取数据包；
                //参数2 A attachment：异步Channel携带的附件，通知回调的时候作为入参使用。即回调方法的第二个参数
                //参数3 CompletionHandler＜Integer,? super A＞：接收通知回调的业务handler，本例程中为ReadCompletionHandler。
                result.read(byteBuffer, byteBuffer, new ReadCompleteHandler(result));
            }
    
            @Override
            public void failed(Throwable exc, SocketServiceCb attachment) {
    
            }
        }
    
        class ReadCompleteHandler implements CompletionHandler<Integer, ByteBuffer> {
    
            private AsynchronousSocketChannel channel;
    
            public ReadCompleteHandler(AsynchronousSocketChannel result) {
                this.channel = result;
            }
    
            @Override
            public void completed(Integer result, ByteBuffer attachment) {
    
                //flip操作，为后续从缓冲区读取数据做准备
                System.out.print("数据大小为："+attachment.remaining());
                byte[] bytes = new byte[attachment.remaining()];
                attachment.get(bytes);
    
                try {
                    System.out.print("服务器接收：" + new String(bytes, "UTF-8"));
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
    
                doWrite("12312");
            }
    
            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                try {
                    this.channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
    
            private void doWrite(String data) {
                ByteBuffer byteBuffer = ByteBuffer.wrap(data.getBytes());
                byteBuffer.flip();
    
                channel.write(byteBuffer, byteBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                    @Override
                    public void completed(Integer result, ByteBuffer attachment) {
                        if (attachment.hasRemaining()) {
    
                            channel.write(attachment, attachment, this);
                        }else {
    
                            latch.countDown();
                        }
                    }
    
                    @Override
                    public void failed(Throwable exc, ByteBuffer attachment) {
    
                        try {
                            channel.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        }
    }

CompletionHandler接口即为回调，它有两个方法，执行成功的回调和异常回调，分别如下。

>  *  public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment)；
>  *  public void failed(Throwable exc, AsyncTimeServerHandler attachment)。

#### AIO的回调模式的客户端代码： ####

CountDownLatch latch;
    
        @Test
        public void testNIOCallBack() throws IOException {
            final AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();
    
            latch = new CountDownLatch(1);
    
            channel.connect(new InetSocketAddress("127.0.0.1", 17777),null, new CompletionHandler() {
    
                @Override
                public void completed(Object result, Object attachment) {
                    System.out.print("链接成功");
                    //final ByteBuffer byteBuffer = ByteBuffer.wrap("12312414".getBytes());
                    byte[] bytes = "1231231231".getBytes();
                    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                    byteBuffer.put(bytes);
    
                    byteBuffer.flip();
    
                    channel.write(byteBuffer, byteBuffer, new CompletionHandler<Integer, ByteBuffer>() {
    
                        @Override
                        public void completed(Integer result, ByteBuffer attachment) {
                            if(attachment.hasRemaining()){
                                channel.write(attachment, attachment, this);
                            }else {
                                System.out.print("发送成功");
                                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                                channel.read(readBuffer, readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                                    @Override
                                    public void completed(Integer result, ByteBuffer attachment) {
    
                                        byte[] bytes = new byte[attachment.remaining()];
                                        attachment.get(bytes);
    
                                        System.out.print("接受信息："+new String(bytes));
    
                                        latch.countDown();
                                    }
    
                                    @Override
                                    public void failed(Throwable exc, ByteBuffer attachment) {
                                        exc.printStackTrace();
    
                                    }
                                });
                            }
                        }
    
                        @Override
                        public void failed(Throwable exc, ByteBuffer attachment) {
                            exc.printStackTrace();
    
                        }
                    });
                }
    
                @Override
                public void failed(Throwable exc, Object attachment) {
                  exc.printStackTrace();
                }
            });
    
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

JDK底层通过线程池ThreadPoolExecutor来执行回调通知，异步回调通知类由sun.nio.ch.AsynchronousChannelGroupImpl实现，它经过层层调用，最终回调com.phei.netty.aio.AsyncTimeClientHandler$1.completed方法，完成回调通知。由此我们也可以得出结论：异步Socket Channel是被动执行对象，我们不需要像NIO编程那样创建一个独立的I/O线程来处理读写操作。对于AsynchronousServerSocket Channel和AsynchronousSocketChannel，它们都由JDK底层的线程池负责回调并驱动读写操作。

### Future模式 ###

Future模式服务器端的代码

public class SocketServiceAIO {
    
        private static ExecutorService executorService;
        private static AsynchronousServerSocketChannel serverSocketChannel;
    
        static class ChannelWorker implements Callable<String> {
            private CharBuffer charBuffer;
            private CharsetDecoder decoder = Charset.defaultCharset().newDecoder();
            private AsynchronousSocketChannel channel;
    
            ChannelWorker(AsynchronousSocketChannel channel) {
                this.channel = channel;
            }
    
            @Override
            public String call() throws Exception {
    
                final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    
                //读取请求
                while (channel.read(byteBuffer).get() != -1) {
                    byteBuffer.flip();
                    charBuffer = decoder.decode(byteBuffer);
                    String request = charBuffer.toString().trim();
                    System.out.println("客户端请求：" + request);
    
                    ByteBuffer outByteBuffer = ByteBuffer.wrap("请求收到".getBytes());
    
                    Future future = channel.write(outByteBuffer);
    
                    future.get();
    
                    if (byteBuffer.hasRemaining()) {
                        byteBuffer.compact();
                    } else {
                        byteBuffer.clear();
                    }
    
                }
                channel.close();
                return "OK";
            }
        }
    
        private static void init() throws IOException {
    
            executorService = Executors.newCachedThreadPool(Executors.defaultThreadFactory());
            serverSocketChannel = AsynchronousServerSocketChannel.open();
    
            if (serverSocketChannel.isOpen()) {
                serverSocketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024);
                serverSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
    
                serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 17777));
            } else {
                throw new RuntimeException("通道未打开");
            }
    
        }
    
        private static void start() {
    
            System.out.println("等待客户端请求...");
    
            while (true) {
                //接收客户端请求
                Future<AsynchronousSocketChannel> future = serverSocketChannel.accept();
    
                try {
    
                    //获取请求
                    AsynchronousSocketChannel channel = future.get();
                    //提交给线程池
                    executorService.submit(new ChannelWorker(channel));
                } catch (Exception ex) {
                    ex.printStackTrace();
                    System.err.println("服务器关闭");
                    executorService.shutdown();
    
                    while (!executorService.isTerminated()) {
    
                    }
                    break;
    
                }
            }
        }
    
        public static void startTCPService() {
            try {
                init();
            } catch (IOException e) {
                e.printStackTrace();
                System.out.println("AIO初始化失败");
            }
    
    
            try {
                start();
    
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println("AIO初始化失败");
            }
        }
    }

Future客户端代码：

@Test
        public void test() {
    
            try {
                start();
    
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println("AIO初始化失败");
            }
        }
    
    
        private void start() throws IOException, ExecutionException, InterruptedException {
            AsynchronousSocketChannel socketChannel = AsynchronousSocketChannel.open();
    
            if (socketChannel.isOpen()) {
                socketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024);
                socketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
    
                Future future = socketChannel.connect(new InetSocketAddress("127.0.0.1", 17777));
    
                Object connect = future.get();
    
                if(connect != null){
                    throw new RuntimeException("链接失败");
                }
    
            } else {
                throw new RuntimeException("通道未打开");
            }
    
            //发送数据
    
            Future future = socketChannel.write(ByteBuffer.wrap("我是客户端".getBytes()));
    
            future.get();
    
            //读取服务器的发送的数据
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    
            //读取服务器
            while(socketChannel.read(byteBuffer).get() != -1){
                byteBuffer.flip();
    
                CharBuffer charBuffer = Charset.defaultCharset().newDecoder().decode(byteBuffer);
    
                System.out.println("服务器说："+charBuffer.toString().trim());
                byteBuffer.clear();
            }
        }

至此，我们学习了在java中几种不同的网络编程模式。这里总是一下：

1.异步非阻塞I/O

即本节所述的JDK 1.7 AIO。很多人喜欢将JDK1.4提供的NIO框架称为异步非阻塞I/O，但是，如果严格按照UNIX网络编程模型和JDK的实现进行区分，实际上它只能被称为非阻塞I/O，不能叫异步非阻塞I/O。因为它是基于Selctor的select/poll模型实现，它是基于I/O复用技术的非阻塞I/O，不是异步I/O。在JDK1.5 update10和Linux core2.6以上版本，Sun优化了Selctor的实现，它在底层使用epoll替换了select/poll，上层的API并没有变化，可以认为是JDK NIO的一次性能优化，但是它仍旧没有改变I/O的模型。

由JDK1.7提供的NIO2.0，新增了异步的套接字通道，它是真正的异步I/O，在异步I/O操作的时候可以传递信号变量，当操作完成之后会回调相关的方法，异步I/O也被称为AIO。

2.多路复用器/选择器 Selector

在前面的章节我们介绍过Java NIO的实现关键是多路复用I/O技术，多路复用的核心就是通过Selector来轮询注册在其上的Channel，当发现某个或者多个Channel处于就绪状态后，从阻塞状态返回就绪的Channel的选择键集合，进行I/O操作。由于多路复用器是NIO实现非阻塞I/O的关键

3.伪异步I/O

伪异步I/O的概念完全来源于实践。在JDK NIO编程没有流行之前，为了解决Tomcat通信线程同步I/O导致业务线程被挂住的问题，大家想到了一个办法：在通信线程和业务线程之间做个缓冲区，这个缓冲区用于隔离I/O线程和业务线程间的直接访问，这样业务线程就不会被I/O线程阻塞。而对于后端的业务侧来说，将消息或者Task放到线程池后就返回了，它不再直接访问I/O线程或者进行I/O读写，这样也就不会被同步阻塞。像这样通过线程池做缓冲区的做法来解决一连接一线程问题，习惯于称它为伪异步I/O，而官方并没有伪异步I/O这种说法，请大家注意。

4.同步阻塞IO

即最简单，也最好理解，一个Socket链接，开启一个线程去处理。

![这里写图片描述][70]

### 如何选择 ###

如果客户端并发连接数不多，服务器的负载也不重，那就完全没必要选择NIO做服务端，毕竟非阻塞的处理IO是比阻塞IO的响应时间要慢一些(涉及多线程的上下文切换等问题)；如果是相反情况，那就要考虑选择合适的NIO框架进行开发。但并建议直接利用JDK的NIO来开发。

开发出高质量的NIO程序并不是一件简单的事情，除去NIO固有的复杂性和BUG不谈，作为一个NIO服务端，需要能够处理网络的闪断、客户端的重复接入、客户端的安全认证、消息的编解码、半包读写、网络拥塞等情况。由于NIO还涉及到Reactor模式，如果你没有足够的NIO网络编程和多线程编程经验积累，一个NIO框架的稳定往往需要半年甚至更长的时间。更为糟糕的是，一旦在生产环境中发生问题，往往会导致跨节点的服务调用中断，严重的可能会导致整个集群环境都不可用，需要重启服务器，这种非正常停机会带来巨大的损失。

从可维护性角度看，由于NIO采用了异步非阻塞编程模型，而且是一个I/O线程处理多条链路，它的调试和跟踪非常麻烦，特别是生产环境中的问题，我们无法进行有效的调试和跟踪，往往只能靠一些日志来辅助分析，定位难度很大。

由于上述原因，在大多数场景下，不建议大家直接使用JDK的NIO类库，除非你精通NIO编程或者有特殊的需求。在绝大多数的业务场景中，我们可以使用NIO框架Netty来进行NIO编程，它既可以作为客户端也可以作为服务端，同时支持UDP和异步文件传输，功能非常强大。

后续笔者将记录Netty的学习笔记

[70]: /images/20220524/d40b719f59f64a70b3f8a34d74e7f7e4.png