高并发网络编程之NIO非阻塞网络编程

程序员文章站 2022-05-05 16:36:14

...

NIO非阻塞网络编程

始于Java1.4，提供了新的JAVA IO操作非阻塞API。用意是替代Java IO和Java Networking相关的API。

三个核心组件：

Buffer缓冲区
Channel通道
Selector选择器

Buffer缓冲区

缓冲区本质上是一个可以写入数据的内存块（类似数组），然后可以再次读取。此内存块包含在NIO Buffer对象中，该对象提供了一组方法，可以更轻松地使用内存块。

相比较直接对数组的操作，Buffer API更加容易操作和管理。

使用Buffer进行数据写入和读取，需要进行如下四个步骤：

将数据写入缓冲区
调用buffer.flip()，转换为读取模式
缓冲区读取数据
调用buffer.clear()或buffer.compact()清除缓冲区

Buffer工作原理：

Buffer三个重要属性：

capacity容量：作为一个内存块，Buffer具有一定的固定大小，也称为容量。
position位置：写入模式时代表写数据的位置。读取模式时代表读取数据的位置。
limit限制：写入模式，限制等于buffer的容量。读取模式下，limit等于写入的数据量。

高并发网络编程之NIO非阻塞网络编程

public class BufferDemo {
    public static void main(String[] args) {
        // 构建一个byte字节缓冲区，容量是4
        ByteBuffer byteBuffer = ByteBuffer.allocate(4);
        // 默认写入模式，查看三个重要的指标
        System.out.println(String.format("初始化：capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));
        // 写入2字节的数据
        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方法，也是可以读取数据的，但是position记录读取的位置不对)
        System.out.println("#######开始读取");
        byteBuffer.flip();
        byte a = byteBuffer.get();
        System.out.println(a);
        byte b = byteBuffer.get();
        System.out.println(b);
        System.out.println(String.format("读取2字节数据后，capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));

        // 继续写入3字节，此时读模式下，limit=3，position=2.继续写入只能覆盖写入一条数据
        // clear()方法清除整个缓冲区。compact()方法仅清除已阅读的数据。转为写入模式
        byteBuffer.compact(); // buffer : 1 , 3
        byteBuffer.put((byte) 3);
        byteBuffer.put((byte) 4);
        byteBuffer.put((byte) 5);
        System.out.println(String.format("最终的情况，capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
                byteBuffer.position(), byteBuffer.limit()));

        // rewind() 重置position为0
        // mark() 标记position的位置
        // reset() 重置position为上次mark()标记的位置

    }
}

ByteBuffer内存类型

ByteBuffer为性能关键型代码提供了直接内存（direct堆外）和非直接内存（heap堆）两种实现。

堆外内存获取方式：

ByteBuffer directByteBuffer = ByteBuffer.allocateDirect(noBytes);

好处：

进行网络IO或者文件IO时比heapBuffer少一次拷贝。（file/socket ---- OS memory ---- jvm heap (可以用卖股票的方式来理解：先将股票转化为现金，再用现金进行交易)）GC会移动对象内存，在写file或socket的过程中，JVM的实现中，会先把数据复制到堆外，再进行写入。
GC范围之外，降低GC压力，但实现了自动管理。DirectByteBuffer中有一个Cleaner对象（PhantomReference），Cleaner被GC前会执行celan方法，触发DirectByteBuffer中定义的Deallocator

建议：

性能确实可观的时候采取使用；分配给大型、长寿命；（网络传输、文件读写场景）
通过虚拟机参数MaxDirectMemorySize限制大小，防止耗尽整个机器的内存；

public class DirectBufferDemo {
	public static void main(String[] args) {
		// 构建一个byte字节缓冲区，容量是4
		ByteBuffer byteBuffer = ByteBuffer.allocateDirect(4);
		// 默认写入模式，查看三个重要的指标
		System.out.println(String.format("初始化：capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
				byteBuffer.position(), byteBuffer.limit()));
		// 写入2字节的数据
		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方法，也是可以读取数据的，但是position记录读取的位置不对)
		System.out.println("#######开始读取");
		byteBuffer.flip();
		byte a = byteBuffer.get();
		System.out.println(a);
		byte b = byteBuffer.get();
		System.out.println(b);
		System.out.println(String.format("读取2字节数据后，capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
				byteBuffer.position(), byteBuffer.limit()));

		// 继续写入3字节，此时读模式下，limit=3，position=2.继续写入只能覆盖写入一条数据
		// clear()方法清除整个缓冲区。compact()方法仅清除已阅读的数据。转为写入模式
		byteBuffer.compact();
		byteBuffer.put((byte) 3);
		byteBuffer.put((byte) 4);
		byteBuffer.put((byte) 5);
		System.out.println(String.format("最终的情况，capacity容量：%s, position位置：%s, limit限制：%s", byteBuffer.capacity(),
				byteBuffer.position(), byteBuffer.limit()));
		byteBuffer.array();
		// rewind() 重置position为0
		// mark() 标记position的位置
		// reset() 重置position为上次mark()标记的位置
	}
}

Channel通道

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Gqtrene3-1589444120550)(images/NIO与BIO通信区别.png)]

和标准IO Stream操作的区别：

在一个通道内进行读取和写入

stream通常是单向的（input或output）

可以非阻塞读取和写入通道

通道始终读取或写入缓冲区

Channel的API涵盖了UDP/TCP网络和文件IO

FileChannel、DatagramChannel、SocketChannel、ServerSocketChannel

SocketChannel

SocketChannel用于建立TCP网络连接，类似java.net.Socket。有两种创建socketChannel形式：

客户端主动发起和服务器的连接。
服务端获取的新连接。

注意：

**write写：**write()在尚未写入任何内容是就可能返回了。需要在循环中调用write()。

**read读：**read()方法可能直接返回而根本不读取任何数据，根据返回的int值判断读取了多少字节。

ServerSocketChannel

ServerSocketChannel可以监听新建的TCP连接通道，类似serverSocket。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-He9xMnqd-1589444120553)(images/ServerSocketChannel.png)]

**serverSocketChannel.accept()：**如果该通道处于非阻塞模式，那么如果没有挂起的连接，该方法将立即返回null。必须检查返回的SocketChannel是否为null。

Selector选择器

Selector是一个Java NIO组件，可以检查一个或多个NIO通道，并确定哪些通道已准备好进行读取或写入。实现单个线程可以管理多个通道，从而管理多个网络连接。

一个线程使用Selector监听多个channel的不同事件：

四个事件分别对应selectionKey四个常量。

Connect连接（SelectionKey.OP_CONNECT）
Accept准备就绪（OP_ACCEPT）
Read读取（OP_READ）
Write写入（OP_WRITE）

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EuOWCSUU-1589444120554)(images/Selector工作流程.png)]

核心概念：

实现一个线程处理多个通道的核心概念理解：事件驱动机制。

非阻塞的网络通道下，开发者通过Selector注册对于通道感兴趣的事件类型，线程通过监听事件来触发相应的代码执行。（拓展：更底层是操作系统的多路复用机制）

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-MAgdiQ0n-1589444120555)(images/Selector核心概念.png)]

NIO对比BIO

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cUN5LhL3-1589444120556)(images/BIO线程模型.png)]

阻塞IO，线程等待时间长
一个线程负责一个连接处理
线程多且利用率低

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Y61abJq6-1589444120556)(images/NIO线程模型.png)]

非阻塞IO，线程利用率更高
一个线程处理多个连接事件
性能更强大

如果程序需要支撑大量的连接，使用NIO是最好的方式。

Tomcat8中，已经完全去除BIO相关的网络处理代码，默认采用NIO进行网络处理。

Reactor网络模型

网络模型：

应用于多线程高并发场景、海量请求，充分利用CPU多核的性能

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vxJF6Mt4-1589444120556)(images/Ractor模式.png)]

/**
 * NIO selector 多路复用reactor线程模型
 */
public class NIOServerV3 {
    /** 处理业务操作的线程 */
    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(8080));
        System.out.println("启动完成，端口8080");
    }

    public static void main(String[] args) throws Exception {
        NIOServerV3 nioServerV3 = new NIOServerV3();
        nioServerV3.newGroup(); // 1、 创建main和sub两组线程
        nioServerV3.initAndRegister(); // 2、 创建serverSocketChannel，注册到mainReactor线程上的selector上
        nioServerV3.bind(); // 3、 为serverSocketChannel绑定端口
    }
}

相关标签： java高性能编程网络 nio java

上一篇：蟑螂送给你的房租

下一篇： Java线程的几种状态分析

高并发网络编程之NIO非阻塞网络编程

NIO非阻塞网络编程

Buffer缓冲区

Buffer工作原理：

ByteBuffer内存类型

Channel通道

SocketChannel

ServerSocketChannel

Selector选择器

NIO对比BIO

Reactor网络模型

javaNIO实战-- java NIO的非阻塞式IO网络编程UDP协议实战

javaNIO实战-- java NIO的非阻塞式IO网络编程UDP协议实战

NIO非阻塞网络编程

高并发网络编程之NIO非阻塞网络编程

NIO网络编程非阻塞

linux网络高级编程之非阻塞