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

文章目录

NIO非阻塞网络编程
- Buffer缓冲区
- - Buffer工作原理：
  - ByteBuffer内存类型
- Channel通道
- - SocketChannel
  - ServerSocketChannel
- Selector选择器
- NIO对比BIO
- Reactor网络模型

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等于写入的数据量。

public class BufferDemo {public static void main(String[] args) {// 构建一个byte字节缓冲区，容量是4ByteBuffer 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 , 3byteBuffer.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字节缓冲区，容量是4ByteBuffer 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通道

和标准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。

**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）

核心概念：

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

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

NIO对比BIO

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

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

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

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

Reactor网络模型

网络模型：

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

/*** 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;@Overridepublic 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() {@Overridepublic 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(() -> {});// 响应结果 200String 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线程, 只负责处理serverSocketChannelfor (int i = 0; i < mainReactorThreads.length; i++) {mainReactorThreads[i] = new ReactorThread() {AtomicInteger incr = new AtomicInteger(0);@Overridepublic 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、 创建ServerSocketChannelserverSocketChannel = ServerSocketChannel.open();serverSocketChannel.configureBlocking(false);// 2、 将serverSocketChannel注册到selectorint 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绑定端口}
}