Java网络编程--NIO

NIO编程

NIO被称为非阻塞IO,它跟BIO不同的地方在于,它如果没有接收到客户端消息的话,可以不阻塞当前服务线程,从而使当前服务线程去继续接收其他客户端线程的请求。

NIO的三大核心组件

  1. ByteBuffer
  2. ServerSocketChannel
  3. Selector

ByteBuffer的使用

//构建一个byte字节缓冲区,默认分配的是堆内存,容量为4,则此时它的limit也为4
        ByteBuffer byteBuffer = ByteBuffer.allocate(4);

//        ByteBuffer byteBuffer = ByteBuffer.allocateDirect(4);//申请堆外内存

        //ByteBuffer的三个重要属性
        System.out.println(String.format("初始化:capacity容量:%s, position位置:%s, limit限制:%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));
        //写入三个字节的byte整数
        byteBuffer.put((byte) 1);
        byteBuffer.put((byte) 2);
        byteBuffer.put((byte) 3);
        System.out.println(String.format("写入3个字节后的属性情况:capacity容量:%s, position位置:%s, limit限制:%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));

        //读取元素时,需要调用flip()方法将下表重新回到0
        System.out.println("现在开始读取元素");
        byteBuffer.flip();
        byte firstByte = byteBuffer.get();
        System.out.println("读取的第一个元素是:" + firstByte);
        byte secondByte = byteBuffer.get();
        System.out.println("读取的第二个元素是:" + secondByte);
        //现在属于读模式的情况,现在已经存放了三个元素,所以只能读到第三个元素,它的下标是2;所以不能读取到的下标limit=3
        System.out.println(String.format("读取2字节数据后,capacity容量:%s, position位置:%s, limit限制:%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));

        //继续对byteBuffer进行写操作
        //此时读模式下,limit=3,position=2.继续写入只能覆盖写入一条数据
        // clear()方法清除整个缓冲区。compact()方法仅清除已阅读的数据。转为写入模式
        System.out.println("切换为写入模式(使用compact方法清除掉已经读取的数据)");
        System.out.println("写入前buffer的内容为:");
        for (int i = 0; i < byteBuffer.limit(); i++) {
            System.out.print(byteBuffer.get(i) + " ");
        }
        byteBuffer.compact();

        byteBuffer.put((byte) 4);
        byteBuffer.put((byte) 5);

        System.out.println();
        System.out.println(String.format("写入2字节数据后,capacity容量:%s, position位置:%s, limit限制:%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));

        System.out.println("写入后buffer的内容为:");
        for (int i = 0; i < byteBuffer.limit(); i++) {
            System.out.print(byteBuffer.get(i) + " ");
        }

        System.out.println();
        //读取里边的元素
        byteBuffer.flip();
        System.out.println("再次切换到读模式");
        System.out.println(String.format("当前byteBuffer,capacity容量:%s, position位置:%s, limit限制:%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));
        System.out.print("当前的数据为:" );
        for (int i = 0; i < byteBuffer.limit(); i++) {
            System.out.print(byteBuffer.get(i) + " ");
        }

ByteBuffer为性能关键型代码提供了直接内存(direct对外)非直接内存(heap堆)两种实现。
使用堆外内存的好处

  1. 进行网络IO或者文件IO时比堆内存少一次拷贝。正常的堆内对象,GC时会移动对象内存,在写file或socket的过程中,会先把数据复制到堆外,再进行写入。
  2. 直接内存在GC范围之外,减低GC压力,但实现了自动管理。DirectByteBuffer中有一个Cleaner对象(PhantomReference),Cleaner被GC前会执行clean方法,触发DirectByteBuffer中定义的Deallocator。

使用场合

  1. 性能确实可观的时候才去使用;分配给大型、长寿命对象;比如网络传输、文件读写场景
  2. 通过虚拟机参数MaxDirectMomorySize限制大小,防止耗尽整个机器的内存

channel通道

相对于BIO来说,NIO编程中不需要以byte数组为数据存储来使用InputStream和OutPutStream两个对象来作为数据的传输对象;NIO编程中可以使用channel对象来进行网络的连接并且以非阻塞的方式通过ByteBuffer为数据载体使用channel来作为传输通道
channel分为服务端的ServerSocketChannel和客户端的SocketChannel
客户端使用如下:

/**
	客户端SocketChannel的使用
*/
public class NIOClient {
    public static void main(String[] args) throws Exception {
    	//获取通道
        SocketChannel socketChannel = SocketChannel.open();
        //设置是否为阻塞IO
        socketChannel.configureBlocking(false);
        //连接端口
        socketChannel.connect(new InetSocketAddress("127.0.0.1", 8080));

        while (!socketChannel.finishConnect()) {
            // 没连接上,则一直等待
            Thread.yield();
        }
        Scanner scanner = new Scanner(System.in);
        System.out.println("请输入:");
        // 发送内容
        String msg = scanner.nextLine();
        ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
        while (buffer.hasRemaining()) {
            socketChannel.write(buffer);
        }
        // 读取响应
        System.out.println("收到服务端响应:");
        ByteBuffer requestBuffer = ByteBuffer.allocate(1024);

        while (socketChannel.isOpen() && socketChannel.read(requestBuffer) != -1) {
            // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
            if (requestBuffer.position() > 0) break;
        }
        requestBuffer.flip();
        byte[] content = new byte[requestBuffer.limit()];
        requestBuffer.get(content);
        System.out.println(new String(content));
        scanner.close();
        socketChannel.close();
    }

}

服务器端由于涉及到Selector的使用,所以我们先来了解一下Selector。

Selector选择器

Selector可以检查一个或多个NIO通道,并确定哪些通道已经准备好进行读取或者写入。实现单个线程可以管理多个通道,从而管理多个网络连接
一个线程使用Selector监听多个channel的不同事件,它包含四类事件,对应SelectionKey的四个常量:

  1. Connect连接(SelectionKey.OP_CONNECT)
  2. Accept准备就绪(SelectionKey.OP_ACCEPT)
  3. Read读取(SelectionKey.OP_READ)
  4. Write写入(SelectionKey.OP_WRITE)

实现一个线程处理多个通道的核心理念:事件驱动机制。非阻塞的网络通道下,开发者通过Selector注册对于通道感兴趣的事件类型,线程通过监听事件来触发相应的代码执行。(更底层是操作系统的多路复用机制)

它的使用如下:

/**
 * 结合selector实现非阻塞的服务端,借助消息机制放弃对channel的轮训
 */
public class NIOServer {

    public static void main(String[] args) throws IOException {
        // 1. 创建网络服务端ServerSocketChannel
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        //设置为非阻塞IO方式
        serverSocketChannel.configureBlocking(false);

        //2.构建一个Selector,将channel注册上去
        Selector selector = Selector.open();
        // 将serverSocketChannel注册到selector
        SelectionKey selectionKey = serverSocketChannel.register(selector, 0, serverSocketChannel);
        //对serverSocketChannel上面的accept事件感兴趣(serverSocketChannel只能支持accept操作)
        selectionKey.interestOps(SelectionKey.OP_ACCEPT);

        //3.绑定服务器端口
        serverSocketChannel.socket().bind(new InetSocketAddress(8000));
        System.out.println("服务器启动");

        //开始轮询
        while(true){
            //使用selector.select方法进行轮询,该方法有阻塞效果,直到有事件通知才会有返回
            selector.select();
            //获取所有的注册事件key
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            //遍历这些注册事件的key
            Iterator<SelectionKey> keyIterator = selectionKeys.iterator();

            while (keyIterator.hasNext()){
                SelectionKey currentSelectionKey = keyIterator.next();
                keyIterator.remove();

                //关注read和accept这两类注册事件
                if(currentSelectionKey.isAcceptable()){
                    //获取当前key对应的事件对象
                    ServerSocketChannel current_serverSocketChannel = (ServerSocketChannel) currentSelectionKey.attachment();
                    //来获取socketChannel,将拿到的客户端连接通道,注册到selector上面
                    //mainReactor
                    SocketChannel clientSocketChannel = current_serverSocketChannel.accept();
                    clientSocketChannel.configureBlocking(false);
                    clientSocketChannel.register(selector,SelectionKey.OP_READ,clientSocketChannel);
                    System.out.println("收到新连接 : " + clientSocketChannel.getRemoteAddress());
                }

                //判断是否有数据可读
                if(currentSelectionKey.isReadable()){
                    SocketChannel socketChannel = (SocketChannel) currentSelectionKey.attachment();
                    try{
                        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        while (socketChannel.isOpen() && socketChannel.read(byteBuffer)!= -1){
                            // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
                            if (byteBuffer.position() > 0) {break;}
                        }
                        //如果没有数据,则进行其他注册事件的处理
                        if(byteBuffer.position() ==0) {continue;}
                        //如果客户端返回有数据,则进行数据读取
                        byteBuffer.flip();
                        //
                        byte[] bytesContent = new byte[byteBuffer.limit()];
                        byteBuffer.get(bytesContent);
                        System.out.println(socketChannel.getRemoteAddress() +"服务器,收到的数据是:" + String.valueOf(bytesContent));
                        // TODO 业务操作 数据库 接口调用等等

                        // 响应结果 200
                        String response = "HTTP/1.1 200 OK\r\n" +
                                "Content-Length: 11\r\n\r\n" +
                                "hello,i am server";
                        ByteBuffer output_buffer = ByteBuffer.wrap(response.getBytes());
                        while (output_buffer.hasRemaining()){
                            socketChannel.write(output_buffer);
                        }
                    }catch (IOException ioException){
                        //出现异常则取消该事件
                        currentSelectionKey.cancel();
                    }
                }
            }
            selector.selectNow();
        }
    }
}

注意:上边代码中一个selector监听所有事件,一个线程处理所有请求事件会成为瓶颈,并且不能充分发挥多核CPU的作用,要结合多线程的使用来处理。这里我们可以参考Doug Lea的《Scalable IO in Java》来把NIO和多线程技术结合起来。

多路复用的Reactor线程模型

如果服务端只使用一个线程来处理所有请求的话,就不能够高效的使用到CPU,所以我们结合NIO和多线程技术来实现一种Reactor的线程模型。
代码如下:

/**
 * NIO selector 多路复用reactor线程模型
 */
public class NIOServer_Reactor {
    /** 处理业务操作的线程 */
    private static ExecutorService workPool = Executors.newCachedThreadPool();

    /**
     * 封装了selector.select()等事件轮询的代码
     */
    abstract class ReactorThread extends Thread {

        Selector selector;
        LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();

        /**
         * Selector监听到有事件后,调用这个方法
         */
        public abstract void handler(SelectableChannel channel) throws Exception;

        private ReactorThread() throws IOException {
            selector = Selector.open();
        }

        volatile boolean running = false;

        @Override
        public void run() {
            // 轮询Selector事件
            while (running) {
                try {
                    // 执行队列中的任务
                    Runnable task;
                    while ((task = taskQueue.poll()) != null) {
                        task.run();
                    }
                    selector.select(1000);

                    // 获取查询结果
                    Set<SelectionKey> selected = selector.selectedKeys();
                    // 遍历查询结果
                    Iterator<SelectionKey> iter = selected.iterator();
                    while (iter.hasNext()) {
                        // 被封装的查询结果
                        SelectionKey key = iter.next();
                        iter.remove();
                        int readyOps = key.readyOps();
                        // 关注 Read 和 Accept两个事件
                        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                            try {
                                SelectableChannel channel = (SelectableChannel) key.attachment();
                                channel.configureBlocking(false);
                                handler(channel);
                                if (!channel.isOpen()) {
                                    key.cancel(); // 如果关闭了,就取消这个KEY的订阅
                                }
                            } catch (Exception ex) {
                                key.cancel(); // 如果有异常,就取消这个KEY的订阅
                            }
                        }
                    }
                    selector.selectNow();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        private SelectionKey register(SelectableChannel channel) throws Exception {
            // 为什么register要以任务提交的形式,让reactor线程去处理?
            // 因为线程在执行channel注册到selector的过程中,会和调用selector.select()方法的线程争用同一把锁
            // 而select()方法是在eventLoop中通过while循环调用的,争抢的可能性很高,为了让register能更快的执行,就放到同一个线程来处理
            FutureTask<SelectionKey> futureTask = new FutureTask<>(() -> channel.register(selector, 0, channel));
            taskQueue.add(futureTask);
            return futureTask.get();
        }

        private void doStart() {
            if (!running) {
                running = true;
                start();
            }
        }
    }

    private ServerSocketChannel serverSocketChannel;
    // 1、创建多个线程 - accept处理reactor线程 (accept线程)
    private ReactorThread[] mainReactorThreads = new ReactorThread[1];
    // 2、创建多个线程 - io处理reactor线程  (I/O线程)
    private ReactorThread[] subReactorThreads = new ReactorThread[8];

    /**
     * 初始化线程组
     */
    private void newGroup() throws IOException {
        // 创建IO线程,负责处理客户端连接以后socketChannel的IO读写
        for (int i = 0; i < subReactorThreads.length; i++) {
            subReactorThreads[i] = new ReactorThread() {
                @Override
                public void handler(SelectableChannel channel) throws IOException {
                    // work线程只负责处理IO处理,不处理accept事件
                    SocketChannel ch = (SocketChannel) channel;
                    ByteBuffer requestBuffer = ByteBuffer.allocate(1024);
                    while (ch.isOpen() && ch.read(requestBuffer) != -1) {
                        // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
                        if (requestBuffer.position() > 0) {break;}
                    }
                    if (requestBuffer.position() == 0) {return;} // 如果没数据了, 则不继续后面的处理
                    requestBuffer.flip();
                    byte[] content = new byte[requestBuffer.limit()];
                    requestBuffer.get(content);
                    System.out.println(new String(content));
                    System.out.println(Thread.currentThread().getName() + "收到数据,来自:" + ch.getRemoteAddress());

                    // TODO 业务操作 数据库、接口...
                    workPool.submit(() -> {
                    });

                    // 响应结果 200
                    String response = "HTTP/1.1 200 OK\r\n" +
                            "Content-Length: 11\r\n\r\n" +
                            "Hello World";
                    ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
                    while (buffer.hasRemaining()) {
                        ch.write(buffer);
                    }
                }
            };
        }

        // 创建mainReactor线程, 只负责处理serverSocketChannel
        for (int i = 0; i < mainReactorThreads.length; i++) {
            mainReactorThreads[i] = new ReactorThread() {
                AtomicInteger incr = new AtomicInteger(0);

                @Override
                public void handler(SelectableChannel channel) throws Exception {
                    // 只做请求分发,不做具体的数据读取
                    ServerSocketChannel ch = (ServerSocketChannel) channel;
                    SocketChannel socketChannel = ch.accept();
                    socketChannel.configureBlocking(false);
                    // 收到连接建立的通知之后,分发给I/O线程继续去读取数据
                    int index = incr.getAndIncrement() % subReactorThreads.length;
                    ReactorThread workEventLoop = subReactorThreads[index];
                    workEventLoop.doStart();
                    SelectionKey selectionKey = workEventLoop.register(socketChannel);
                    selectionKey.interestOps(SelectionKey.OP_READ);
                    System.out.println(Thread.currentThread().getName() + "收到新连接 : " + socketChannel.getRemoteAddress());
                }
            };
        }


    }

    /**
     * 初始化channel,并且绑定一个eventLoop线程
     *
     * @throws IOException IO异常
     */
    private void initAndRegister() throws Exception {
        // 1、 创建ServerSocketChannel
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        // 2、 将serverSocketChannel注册到selector
        int index = new Random().nextInt(mainReactorThreads.length);
        mainReactorThreads[index].doStart();
        SelectionKey selectionKey = mainReactorThreads[index].register(serverSocketChannel);
        selectionKey.interestOps(SelectionKey.OP_ACCEPT);
    }

    /**
     * 绑定端口
     *
     * @throws IOException IO异常
     */
    private void bind() throws IOException {
        //  1、 正式绑定端口,对外服务
        serverSocketChannel.bind(new InetSocketAddress(8000));
        System.out.println("启动完成,端口8000");
    }

    public static void main(String[] args) throws Exception {
        NIOServer_Reactor nioServer_Reactor = new NIOServer_Reactor();
        nioServer_Reactor.newGroup(); // 1、 创建main和sub两组线程
        nioServer_Reactor.initAndRegister(); // 2、 创建serverSocketChannel,注册到mainReactor线程上的selector上
        nioServer_Reactor.bind(); // 3、 为serverSocketChannel绑定端口
    }
}
上一篇:Java NIO?看这一篇就够了!


下一篇:异步文件通道Java NIO你需要了解多少,来看看这篇文章