Netty系列整体栏目
深入理解NIO的基本原理和底层实现
一，深入理解NIO的底层原理
在上一篇中，讲解了bio的底层原理和具体实现，虽然bio在一定场景下也可以进行通信，但是随着互联网越来越多业务的场景，bio会存在阻塞的弊端被暴露无疑，在并发量稍微大点的地方，通过bio实现的网络编程会显得略显吃力。于是在jdk1.4之后，引入了一个新东西 NIO ，由于bio原名叫做 Blocking IO阻塞io，因此新网络编程的取名nio，有着 NoBlocking IO即不阻塞io，当然也有的地方取名为new io。
在讲解nio之前，依旧和以前的学习一样，不能脱离官网进行学习：netty官网地址 ，用户指南可以参考4.1版本
1，Reactor反应堆模式
网络编程从bio废弃到再到nio的崛起，跟nio的底层实现有着很大的联系，其最主要的设计思想就是这个 reactor 反应堆模式，总结这个reactor模式主要有三点：注册感兴趣的事件，扫描是否有感兴趣的事件发生，在事件发生之后做出相应的处理
在讲解这个反应堆模式之前，先通过一个生活中的案例来讲述这个事情，以我们去线下餐厅点餐为例，首先用户扫码点餐，然后点餐系统返回一个排队的号码，再服务员喊到号码的时候去取餐，就以这个案例来说明一下什么是反应堆模式。
1.1，通过餐厅描述Bio
首先在上面的这个点餐的案例中，bio的实现如下，就有点类似于用户直接和厨师直接进行交流，告诉厨师要什么菜，当没有用户点餐时，那么厨师就会一直等待用户来点餐，直到有用户点餐为止，如果一直没有用户点餐，那么厨师就会一直处于阻塞状态进行等待，这里就对应了bio服务端，没有客户端请求时会长期处于一个阻塞状态；
如果已经有了一个用户点餐，那么厨师会先炒这个用户的菜，当有其他用户来点餐时，那么其他用户会处于阻塞状态，只有等帮第一个用户炒完菜之后，厨师才能和第二个用户进行交流，第二个用户才能把自己需要什么菜告诉厨师，在如果在上一个用户的菜还没炒玩之前，那么下一个用户则会处于一个阻塞等待状态。因此这样效率肯定是非常低下的，那么毫无疑问，bio的这种方式注定是要被淘汰的。(这里服务端默认为在一个cpu里面，就是说一个cpu中只有一个线程去处理请求，上面案例对应的就是服务端对应的就是一个厨师，厨师就是老板，其他顾客就是对应的服务端)
1.2，通过餐厅引入nio
由于一个厨师对应多个用户效率会十分的低下，而且如果用户量稍微大一点，那么每个用户就不用去干其他的事情，就一直排队阻塞在那里，因此严重的影响整个系统的吞吐量以及严重的影响用户的体验感。随着客户的增加，或者午餐这段高峰期，为了解决用户长时间等待问题，那么就可以做一个点餐系统，用户只需扫码点餐即可，当用户点餐完成之后，可以去做用户自己想做的事情，如出去逛逛等，此时系统会给用户一个点餐号，此时就解决了用户长时间排队阻塞的问题。厨师这边也不需要每次只处理一个请求，如多个用户点同一个菜，那么厨师可以一次性炒多份菜，这样也提高了厨师这边的效率。当厨师将菜炒好之后，只需要服务员通过念号或者通过公众号通知订餐的用户即可。
反应堆模式就是，不能一直等着客户端去等待服务端的响应，而是通过某个中间层，客户端先向中间层注册一个事件，当服务端有空做出响应的时候再通过定时任务去扫描这个中间层，当中间层发现有注册的事件之后，再去通知客户端，这样就可以减少客户端的等待时间。换句话就是说，通过请求响应的模式来说，客户端向服务端发送一个请求之后，如果服务端长时间没有响应，那么客户端可以结束此次请求，服务端来不及响应，但是服务端得记录这个请求的记录，当服务端有空的时候，再去扫描这个记录，再去响应这个请求，再通过通知异步的去响应对应的请求。
2，NIO三大核心组件
在nio编程中，里面有三大核心组件，分别是 Selector、Channel、Buffer 三大组件。
在上面讲解了通过餐厅系统去了解nio的内部实现，在这三大组件中，扮演的角色分别如下：
- 由于在网络编程中，基本是基于tcp协议去实现客户端和服务端之间的通信，因此通过socket将tcp协议封装，而这里的channel，是对这个socket进行了再次的封装。也就是说，只需要创建这个channel实例就可以完成双端之间的通信，因此点餐系统里面的用户和厨师之间的交流就是通过这个channel去实现的，那么channel扮演的角色就是完成客户和厨师之间的最终交流
- Selector就是一个选择器，通过这个餐厅系统，可以发现引入了一个新的点餐系统，用于注册客户的订单以及在订单完成之后给予响应，就是通知下单的客户，因此这个Selector选择器扮演的角色就是这个点餐系统，也是这个反应堆模式的核心，用于注册客户端事件，扫描这些注册的事件，并对这些事件做出具体的响应
- 而这个Buffer，就是nio和bio之间的重大区别，因为这个Buffer就是一个Nio的一个重要的特性，用于面向缓冲流进行编程，这个Buffer指的是应用层之间的buffer，就是已经建立好连接之后，在服务端内部的一个缓冲区，如在这个餐厅系统中，在准备食材的时候也是需要大量时间的，如果先点餐的用户需要准备的食材要久一些，那么厨师可以优先炒后面用户下的单，那么这个Buffer就起到重要的作用了。由于这个bio是串行执行，那么就不存在这个Buffer的说法，但是在这个nio里面，通过这个Buffer让整个系统更加的灵活，即使先建立的请求，也可以后响应，从而提高整个系统的吞吐量。还有比如说可以重复的读取数据，来不及处理的优先放在这个buffer缓冲区，某个buffer缓冲区如果字节数没达到要求可以先去处理其他的缓冲区等，主要是让整个系统更加的灵活多变，从而提高整个系统的吞吐量和响应。同时也是与BIO最大的差异化之一
3，NIO通信原理
通过上面的餐厅事例和讲解NIO内部的三大组件，接下来通过一个发送和接收数据的事例讲解NIO底层到底是如何进行网络通信和数据传输的。
- 首先客户端先向服务端发送一个请求，然后服务端在接收到这个请求之后，服务端首先会通过这个Selector先向本地注册一个连接事件，然后再扫描Channel事件列表，查看是否有感兴趣的Channel事件
- 在Channel中找到这个对连接感兴趣的事件之后，随后通知这个感兴趣的事件，创建一个ServerSocketChannel对象，用于服务端和客户端通过三次握手建立可靠的连接
- 完成建立连接之后，又会去Selector中扫描是否有对读数据感兴趣的事件，如果找到有服务端对读数据感兴趣的事件，又会通知对这个事件感兴趣的具体事件，用于实例化SocketChannel对象，这里的SocketChannel就是建立好连接的Socket对象，用于真正的去读取数据以及发送数据
- socket读取的数据并不是发送给服务端的应用程序，而是将数据先存入到Buffer中，让应用程序去读取buffer里面的数据，从而提高整个架构的吞吐量和效率
- 最后将要响应的数据也存到Buffer中，然后通过感兴趣的写事件，将数据返回给对应的客户端即可
4，通过NIO实现简单网络编程
上面讲解了大量的理论，接下来通过具体的编码，来讲述NIO的底层到底是怎么实现的。首先创建一个服务端的线程，用于接收客户端的请求以及内部做出的响应，接下来创建一个 NioServerTask 的任务类，并且实现一个 Runnable 方法，在该方法中去创建 selector，ServerSocketChannel，SockerChannel、Buffer等对象
package com.zhs.netty.nio.nio;
import com.zhs.netty.nio.Const;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
/**
* 服务端线程具体代码实现
*/
@Slf4j
public class NioServerTask implements Runnable{
private volatile boolean started;
private ServerSocketChannel serverSocketChannel;
private Selector selector;
/**
* 构造方法
* @param port 指定要监听的端口号
*/
public NioServerTask(int port) {
try {
//创建一个选择器
selector = Selector.open();
//创建ServerSocketChannel的实例
serverSocketChannel = ServerSocketChannel.open();
//通道实例设置为非阻塞模式
serverSocketChannel.configureBlocking(false);
//绑定端口
serverSocketChannel.socket().bind(new InetSocketAddress(port));
//注册事件到selector之上，监听客户端连接
serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);
started = true;
log.info("服务器已启动，端口号：" + port);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
while(started){
try {
//selector每隔1s被唤醒一次
selector.select(1000);
//获取全部已经注册的本地事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while(iterator.hasNext()){
SelectionKey key = iterator.next();
//将处理过的本地注册事件给删除
iterator.remove();
handleInput(key);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
//处理具体的事件
private void handleInput(SelectionKey key) throws IOException {
if(key.isValid()){
//处理新接入的客户端的请求
if(key.isAcceptable()){
//获取channels全部事件中对此感兴趣的事件
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
//获取到感兴趣的事件之后，创建一个socket实例，用于发送和读取数据
SocketChannel sc = ssc.accept();
//设置为非阻塞
sc.configureBlocking(false);
//注册一个感兴趣的读事件
sc.register(selector,SelectionKey.OP_READ);
}
//处理对端的发送的数据
if(key.isReadable()){
SocketChannel sc = (SocketChannel) key.channel();
//创建ByteBuffer，开辟一个缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
int readBytes = sc.read(buffer);
if(readBytes>0){
//缓冲区中存在指针，记录有效位置
buffer.flip();
//根本有效位置的指针处创建字节数组
byte[] bytes = new byte[buffer.remaining()];
//将缓冲区可读字节数组复制到新建的数组中
buffer.get(bytes);
String message = new String(bytes,"UTF-8");
log.info("服务器收到消息：" + message);
String result = Const.response(message);
doWrite(sc,result);
}else if(readBytes<0){
//将channels集合的数据取消
key.cancel();
sc.close();
}
}
}
}
/*发送应答消息*/
private void doWrite(SocketChannel sc,String response) throws IOException {
byte[] bytes = response.getBytes();
ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
buffer.put(bytes);
buffer.flip();
sc.write(buffer);
}
}
从上面的代码中可以发现，在服务端中只关注了读的事件，并没有关注写的事件。并且在这个Buffer中，存在一个指针，用于记录buffer的有效位置，这样在读数据时，只需要读取到有效的数据即可。
服务端代码写好之后，接下来编写客户端的代码，代码和客户端基本一样，但是由于客户端不需要提供服务，因此在客户端这边是不需要 ServerSocketChannel 这个组件的。其他的 SocketChannel，Selector，Buffer 还是需要的
package com.zhs.netty.nio.nio;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
/**
* @author zhenghuisheng
* nio客户端请求
*/
@Slf4j
public class NioClientTask implements Runnable{
private String host;
private int port;
private volatile boolean started;
private Selector selector;
private SocketChannel socketChannel;
public NioClientTask(String ip, int port) {
this.host = ip;
this.port = port;
try {
//创建选择器的实例
selector = Selector.open();
//创建ServerSocketChannel的实例
socketChannel = SocketChannel.open();
//设置通道为非阻塞模式
socketChannel.configureBlocking(false);
started = true;
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
try{
doConnect();
}catch(IOException e){
e.printStackTrace();
System.exit(1);
}
//循环遍历selector
while(started){
try{
//无论是否有读写事件发生，selector每隔1s被唤醒一次
selector.select(1000);
//获取全部已经注册的本地事件
Set<SelectionKey> keys = selector.selectedKeys();
//转换为迭代器
Iterator<SelectionKey> it = keys.iterator();
SelectionKey key = null;
while(it.hasNext()){
key = it.next();
it.remove();
try{
handleInput(key);
}catch(Exception e){
if(key != null){
key.cancel();
if(key.channel() != null){
key.channel().close();
}
}
}
}
}catch(Exception e){
e.printStackTrace();
System.exit(1);
}
}
//selector关闭后会自动释放里面管理的资源
if(selector != null)
try{
selector.close();
}catch (Exception e) {
e.printStackTrace();
}
}
//具体的事件处理方法
private void handleInput(SelectionKey key) throws IOException{
if(key.isValid()){
//获得关心当前事件的channel
SocketChannel sc = (SocketChannel) key.channel();
//连接事件
if(key.isConnectable()){
if(sc.finishConnect()){
socketChannel.register(selector,
SelectionKey.OP_READ);}
else System.exit(1);
}
//有数据可读事件
if(key.isReadable()){
//创建ByteBuffer，并开辟一个1M的缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
//读取请求码流，返回读取到的字节数
int readBytes = sc.read(buffer);
//读取到字节，对字节进行编解码
if(readBytes>0){
//将缓冲区当前的limit设置为position,position=0，
// 用于后续对缓冲区的读取操作
buffer.flip();
//根据缓冲区可读字节数创建字节数组
byte[] bytes = new byte[buffer.remaining()];
//将缓冲区可读字节数组复制到新建的数组中
buffer.get(bytes);
String result = new String(bytes,"UTF-8");
log.info("客户端收到消息：" + result);
}
//链路已经关闭，释放资源
else if(readBytes<0){
key.cancel();
sc.close();
}
}
}
}
private void doWrite(SocketChannel channel,String request)
throws IOException {
//将消息编码为字节数组
byte[] bytes = request.getBytes();
//根据数组容量创建ByteBuffer
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
//将字节数组复制到缓冲区
writeBuffer.put(bytes);
//flip操作
writeBuffer.flip();
//发送缓冲区的字节数组
channel.write(writeBuffer);
}
private void doConnect() throws IOException{
//非阻塞的连接,这里需要注意，因为客户端和服务端都是无阻塞的，因此可能在三次握手建立连接之前，
//这段注册读的代码就已经走完了，因此在else中增加一个注册连接的代码
if(socketChannel.connect(new InetSocketAddress(host,port))){
socketChannel.register(selector,SelectionKey.OP_READ);
}else{
socketChannel.register(selector,SelectionKey.OP_CONNECT);
}
}
//写数据对外暴露的API
public void sendMsg(String msg) throws Exception{
doWrite(socketChannel, msg);
}
}
接下来进行一个数据的测试，先创建一个服务端的Main方法，然后启动这个Main方法，并且设置端口号为8881
public class NioServer {
private static NioServerTask nioServerTask;
public static void main(String[] args){
nioServerTask = new NioServerTask(8881);
new Thread(nioServerTask,"NioServer").start();
}
}
再创建一个客户端的Main方法，ip设置成本地，端口号设置成服务端设置的端口号
/**
* @author zhenghuisheng
*/
public class NioClient {
private static NioClientTask nioClientTask;
public static void main(String[] args) throws Exception {
nioClientTask = new NioClientTask("127.0.0.1",8881);
new Thread(nioClientTask,"nioClient").start();
//控制台输入
Scanner scanner = new Scanner(System.in);
String message = scanner.next();
while(!StringUtils.isEmpty(message)){
nioClientTask.sendMsg(message);
}
}
}
客户端发送消息：
132432
21:58:41.118 [nioClient] INFO com.zhs.netty.nio.nio.NioClientTask - 客户端收到消息：Hello,132432,Now is Sat May 04 21:58:41 CST 2024
服务端接收到的消息：
21:58:30.767 [main] INFO com.zhs.netty.nio.nio.NioServerTask - 服务器已启动，端口号：8881
21:58:41.114 [NioServer] INFO com.zhs.netty.nio.nio.NioServerTask - 服务器收到消息：132432
到此为止，通过NIO的方式将服务端发送消息和客户端接收消息的代码实现