java NIO非阻塞式IO网络编程学习笔记

（一）NIO

① 介绍

java.nio全称java non-blocking IO（实际上是 new io），是指JDK 1.4 及以上版本里提供的新api（New IO），为所有的原始类型（boolean类型除外）提供缓存支持的数据容器，使用它可以提供非阻塞式的高伸缩性网络。
HTTP2.0使用了多路复用的技术，做到同一个连接并发处理多个请求，而且并发请求的数量比HTTP1.1大了好几个数量级。

② 三大核心组件

高性能网络编程的基础组件， buffer 缓存区、Channel 通道、Selector 选择器。

（二） Buffer缓存区

① 介绍

缓存区本质上是一个可以写入数据的内存块（类似数组），然后可以再次读取。此内存块包含在NIO Buffer 对象中，该对象提供了一组方法，可以更轻松地使用内存块。
相比较直接对数组的操作。Buffer API 更加容易操作和管理。

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

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

③ Buffer工作原理

BUffer三个重要属性，通过完成了数组的封装。

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

④ 源码

import java.nio.ByteBuffer;
import java.nio.IntBuffer;
import java.nio.LongBuffer;

public class BufferDemo {
 public static void main(String[] args) {
 // 构建一个byte字节缓冲区，容量是4
 //堆内存
 ByteBuffer byteBuffer = ByteBuffer.allocate(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(); // 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前会执行clean 方法，触发DirectByteBuffer 中定义Deallocator

建议

性能确实可观的时候才去使用，分配给大型，长寿命（网络传输，文件读写场景）
通过虚拟机参数MaxDirectMemorySize限制大小，防止耗尽整个机器的内存，在JVM之外的内存无法监控。

（三）Channel 通道

① 介绍

Channel的API 涵盖了UDP、TCP网络和文件IO，FileChannel，DatagramChannel，Socket channel ，ServerSocketChannel。

② 和标准IO Stream操作的区别

在一个通道内进行读取和写入stream通常是单向的（input 或 output），可以非堵塞读取和写入通道，通道事中读取或写入缓冲区。

③ SocketChannel

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

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

write写

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

read读

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

④ ServerSocketChannel

ServerSocketChannel 可能监听新建立的TCP连接通道，类似ServerSocket。

ServerSocketChannel.accepta()

如果该通道处于飞度赛模式，那么如何没有挂起的连接，该方法将立即返回null。必须检查返回的SocketChannel是否为null。

⑤ 源码

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Scanner;

public class NIOClient {

 public static void main(String[] args) throws Exception {
 SocketChannel socketChannel = SocketChannel. open ();
 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();
 }

}

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util. iterator ;

/**
 * 直接基于非阻塞的写法,一个 线程 处理 轮询 所有请求
 */
public class NIOServer1 {
 /**
 * 已经建立连接的集合
 */
 private static ArrayList<SocketChannel> channels = new ArrayList<>();

 public static void main(String[] args) throws Exception {
 // 创建网络服务端
 ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
 serverSocketChannel.configureBlocking(false); // 设置为非阻塞模式
 serverSocketChannel.socket(). bind (new InetSocketAddress(8080)); // 绑定端口
 System.out.println("启动成功");
 while (true) {
 SocketChannel socketChannel = serverSocketChannel. accept (); // 获取新tcp连接通道
 // tcp请求 读取/响应
 if (socketChannel != null) {
 System.out.println("收到新连接 : " + socketChannel.getRemoteAddress());
 socketChannel.configureBlocking(false); // 默认是阻塞的,一定要设置为非阻塞
 channels.add(socketChannel);
 } else {
 // 没有新连接的情况下,就去处理现有连接的数据,处理完的就删除掉
 Iterator<SocketChannel> iterator = channels.iterator();
 while (iterator.hasNext()) {
 SocketChannel ch = iterator.next();
 try {
 ByteBuffer requestBuffer = ByteBuffer.allocate(1024);

 if (ch.read(requestBuffer) == 0) {
 // 等于0,代表这个通道没有数据需要处理,那就待会再处理
 continue;
 }
 while (ch.isOpen() && ch.read(requestBuffer) != -1) {
 // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
 if (requestBuffer.position() > 0) break;
 }
 if(requestBuffer.position() == 0) continue; // 如果没数据了, 则不继续后面的处理
 requestBuffer.flip();
 byte[] content = new byte[requestBuffer.limit()];
 requestBuffer.get(content);
 System.out.println(new String(content));
 System.out.println("收到数据,来自：" + ch.getRemoteAddress());

 // 响应结果 200
 String response = "HTTP/1.1 200 OKrn" +
 "Content-Length: 11rnrn" +
 "Hello World";
 ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
 while (buffer.hasRemaining()) {
 ch.write(buffer);
 }
 iterator.remove();
 } catch (IOException e) {
 e.printStackTrace();
 iterator.remove();
 }
 }
 }
 }
 // 用到了非阻塞的API, 再设计上,和BIO可以有很大的不同
 // 问题: 轮询通道的方式,低效,浪费CPU
 }
}

（四）Select选择器

① 介绍

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

② selector 监听多个 channel的不同事件

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

③ selector 选择器

一个线程处理多个通道的核心概念理解：事件驱动机制。
非堵塞的网络通道下，开发者通过Selector注册对于通道感兴趣的事件类型，线程通过监听事件来触发响应的代码执行（最底层hi操作系统的多路复用机制）

④ 源码

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 结合Selector实现的非阻塞服务端(放弃对channel的轮询,借助消息通知机制)
 */
public class NIOServerV2 {

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

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

 // 3. 绑定端口
 serverSocketChannel.socket().bind(new InetSocketAddress(8080));

 System.out.println("启动成功");

 while (true) {
 // 不再轮询通道,改用下面轮询事件的方式.select方法有阻塞效果,直到有事件通知才会有返回
 selector.select();
 // 获取事件
 Set<SelectionKey> selectionKeys = selector.selectedKeys();
 // 遍历查询结果e
 Iterator<SelectionKey> iter = selectionKeys.iterator();
 while (iter.hasNext()) {
 // 被封装的查询结果
 SelectionKey key = iter.next();
 iter.remove();
 // 关注 Read 和 Accept两个事件
 if (key.isAcceptable()) {
 ServerSocketChannel server = (ServerSocketChannel) key.attachment();
 // 将拿到的客户端连接通道,注册到selector上面
 SocketChannel clientSocketChannel = server.accept(); // mainReactor 轮询accept
 clientSocketChannel.configureBlocking(false);
 clientSocketChannel.register(selector, SelectionKey.OP_READ, clientSocketChannel);
 System.out.println("收到新连接 : " + clientSocketChannel.getRemoteAddress());
 }

 if (key.isReadable()) {
 SocketChannel socketChannel = (SocketChannel) key.attachment();
 try {
 ByteBuffer requestBuffer = ByteBuffer.allocate(1024);
 while (socketChannel.isOpen() && socketChannel.read(requestBuffer) != -1) {
 // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
 if (requestBuffer.position() > 0) break;
 }
 if(requestBuffer.position() == 0) continue; // 如果没数据了, 则不继续后面的处理
 requestBuffer.flip();
 byte[] content = new byte[requestBuffer.limit()];
 requestBuffer.get(content);
 System.out.println(new String(content));
 System.out.println("收到数据,来自：" + socketChannel.getRemoteAddress());
 // TODO 业务操作 数据库 接口调用等等

 // 响应结果 200
 String response = "HTTP/1.1 200 OKrn" +
 "Content-Length: 11rnrn" +
 "Hello World";
 ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
 while (buffer.hasRemaining()) {
 socketChannel.write(buffer);
 }
 } catch (IOException e) {
 // e.printStackTrace();
 key.cancel(); // 取消事件订阅
 }
 }
 }
 selector.selectNow();
 }
 // 问题: 此处一个selector监听所有事件,一个线程处理所有请求事件. 会成为瓶颈! 要有多线程的运用
 }
}

⑤ NIO 和 BIO 的区别

⑥ NIO Reactor的方式

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 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 OKrn" +
 "Content-Length: 11rnrn" +
 "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绑定端口
 }
}

PS：NIO为开发者提供了功能丰富及强大的IO处理API，但是在应用开发的过程中，直接使用JDK提供的API，比较繁琐，而且要想将性能进行提升，光有NIO还是不够的，还需要将多线程技术与之结合起来。因为网络编程本身的复杂性，以及JDK API开发的使用难度较高，所以开源社区中，涌出来很多的JDK NIO进行封装了，增强后的网络编程框架，例如：Netty、Mina等。

智云一二三科技

java NIO非阻塞式IO网络编程学习笔记

关于作者: 智云科技

给这篇文章的作者打赏

关于作者: 智云科技

相关文章

Map和List的几种遍历方式

全网大佬都在用的Java+Python这两套视频学习教程，学习很重要

JavaWeb快速进阶全套教程(程序员必备2020版)：视频+笔记+源码

热门文章

1分享新浪图床上传接口源码

2PHP简单实现路由Route功能

3Tideways、xhprof 和 xhgui 打造 PHP 非侵入式监控平台

4centos系统如何查看是否安装了mysql

5curl 工具简述