《Netty核心代码解析》JavaNIO详解

---

---

前言

回顾一下Netty的核心API

---

一、Java NIO介绍

关于BIO、NIO的一些介绍，在手写实现人脸识别服务器中有介绍。
NIO的API三大核心：Selector、Channel、Buffer。

（1）每个channel都会对应一个 Buffer
（2） Selector 对应一个线程， 一个线程对应多 个chamel(连接)
（3）该图反应了 有三个channel注册到该seletor /程序
（4）程序切换到哪个channel是有事件决定的，Event就是一个重要的概念
（5）Selector 会根据不同的事件，在各个通道上切换
（6）Buffer 就是个内存块，底层是有一个数组数据的读取写入是通过Buffer,这个和BIO, BIO中要么是输入流，或者是输出流，不能双向，但是NIO的Buffer是可以读也可以写，需要flip 方法切换channel是双向的，可以返回底层操作系统的情况，比如Linux，底层的操作 系统通道就是双向的.
一个线程维护一个Selector，当有客户端请求时，将socketchannel注册到selector上，监听客户端的事件。

二、 缓冲区

缓冲区（Buffer）：缓冲区本质上是一个可以读写数据的内存块，可以理解成是一个容器对象(含数组)，该对象提供了一组方法，可以更轻松地使用内存块，，缓冲区对象内置了一些机制，能够跟踪和记录缓冲区的状态变化情况。Channel提供从文件、网络读取数据的渠道，但是读取或写入的数据都必须经由Buffer.

public abstract class Buffer {

    /**
     * The characteristics of Spliterators that traverse and split elements
     * maintained in Buffers.
     */
    static final int SPLITERATOR_CHARACTERISTICS =
        Spliterator.SIZED | Spliterator.SUBSIZED | Spliterator.ORDERED;

    // Invariants: mark <= position <= limit <= capacity
    private int mark = -1;
    private int position = 0;
    private int limit;
    private int capacity;

public class TestBuffer {
    public static void main(String[] args) {
        IntBuffer buffer = IntBuffer.allocate(1024);
        for (int i = 0; i < buffer.capacity(); i++) {
            buffer.put(i);
        }
        // flip的操作
        // limit = position
        // position = 0
        // mark = -1;
        buffer.flip();
        while (buffer.hasRemaining()) {
            //get后面维护了一个指针，每次get都会往后移动
            System.out.println(buffer.get());
        }
    }
}

一开始position等于0，代表要往0处添加元素，limit为1024，代表写入的边界，capacity为1024，代表缓冲区的容量。后来写入1024个字节之后，position等于1024，limit为1024，代表写入的边界，capacity为1024。然后执行完flip之后，limit赋值为position，相当于记录上一次写入的位置，position赋值为0，因为从0到原来的position位置为写入的值。

二、 Channel

1、NIO的通道类似于流，但有些区别如下：
（1）通道可以同时进行读写，而流只能读或者只能写
（2）通道可以实现异步读写数据通道可以从缓冲读数据，也可以写数据到缓冲:
2、BIO中的stream是单向的，例如FileInputStream对象只能进行读取数据的操作，而NIO中的通道(Channel)是双向的，可以读操作，也可以写操作。
3、Channel在NIO中是一个接口publicinterfaceChannelextendsCloseable{}
4、常用的Channel类有：FileChannel、DatagramChannel、ServerSocketChannel和SocketChannel

6、FileChannel用于文件的数据读写，DatagramChannel用于UDP的数据读写，ServerSocketChannel和SocketChannel用于TCP的数据读写。

2.1 FileChannel

1、FileChannel的API测试

public class FileChannelTest {
    public static void read() throws IOException {
        FileInputStream fileInputStream = new FileInputStream(new File("1.txt"));
        FileChannel channel = fileInputStream.getChannel();
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        channel.read(byteBuffer);//读到缓冲区内
        byteBuffer.flip();// 反转
        System.out.println("1.txt读取的：" + new String(byteBuffer.array()));
        fileInputStream.close();
    }

    public static void write() throws IOException {
        System.out.println("Hello FileChannel写入1.txt");
        FileOutputStream fileOutputStream = new FileOutputStream(new File("1.txt"));
        String str = "Hello FileChannel";
        FileChannel fileChannel = fileOutputStream.getChannel();
        // Channel是跟Buffer一一对应的，要想操作Channel的读写得创建Buffer
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        buffer.put(str.getBytes());
        // 要开始写入了，需要对Buffer进行反转
        buffer.flip();
        // 这个API拗口，将缓冲区的内容写出去
        fileChannel.write(buffer);
        fileOutputStream.close();
    }

    public static void main(String[] args) {
        try {
            write();
            read();
        } catch (Exception e) {

        }
    }
}

2、复制文件的小例子

public class CopyFileChannelTest {

    public static void main(String[] args) throws IOException {
        // 创建文件输入流，并得到输入通道
        FileInputStream fileInputStream = new FileInputStream("1.txt");
        FileOutputStream fileOutputStream = new FileOutputStream("2.txt");
        // 创建文件输出流，并得到输出通道
        FileChannel inputChannel = fileInputStream.getChannel();
        FileChannel outputChannel = fileOutputStream.getChannel();
        ByteBuffer inbuffer = ByteBuffer.allocate(1024);
        while (true) {
            inbuffer.clear(); // 每次用完重新归位
            int read = inputChannel.read(inbuffer); //读取数据到buffer
            if (read == -1) break;
            inbuffer.flip();//读写切换的时候记得归位
            outputChannel.write(inbuffer);//将缓冲区数据写到输出通道。
        }
        fileInputStream.close();
        fileOutputStream.close();
    }
}

3、transfomer直接拷贝。

 // 创建文件输入流，并得到输入通道
 FileInputStream fileInputStream = new FileInputStream("1.txt");
 FileOutputStream fileOutputStream = new FileOutputStream("2.txt");
 // 创建文件输出流，并得到输出通道
 FileChannel inputChannel = fileInputStream.getChannel();
 FileChannel outputChannel = fileOutputStream.getChannel();
 outputChannel.transferFrom(inputChannel,0,inputChannel.size());

4、ByteBuffer支持类型化的put和get,put放入的是什么数据类型，get就应该使用相应的数据类型来取出，否则可能有BufferUnderflowException异常。

//创建一个Buffer
ByteBufferbuffer=ByteBuffer.allocate(64);//类型化方式放入数据buffer.putInt(100);
buffer.putLong(9);
buffer.putChar('尚');
buffer.putShort((short)4);//取出buffer.flip();
System.out.println();
System.out.println(buffer.getInt());
System.out.println(buffer.getLong());
System.out.println(buffer.getChar());
System.out.println(buffer.getShort());

5、可以将一个普通Buffer转成只读BufferByteBufferreadOnlyBuffer=buffer.asReadOnlyBuffer();

ByteBuffer readOnlyBuffer=inbuffer.asReadOnlyBuffer();
System.out.println(readOnlyBuffer.getClass());

sout:

class java.nio.HeapByteBufferR

6、NIO还提供了MappedByteBuffer，可以让文件直接在内存（堆外的内存）中进行修改，而如何同步到文件由NIO来完成。
建立磁盘文件到内存的映射，可以实现修改映射内的内容。

RandomAccessFilerandomAccessFile=newRandomAccessFile("1.txt","rw");//获取对应的通道
FileChannelchannel=randomAccessFile.getChannel();
/***
	参数1:FileChannel.MapMode.READ_WRITE使用的读写模式
	参数2：0：可以直接修改的起始位置
	参数3:5:是映射到内存的大小(不是索引位置),即将1.txt的多少个字节映射到内存*可以直接修改的范围就是0-5*实际类型DirectByteBuffer*/
	MappedByteBuffer mappedByteBuffer = channel.map(FileChannel.MapMode.READ_WRITE,0,5);
	mappedByteBuffer.put(0,(byte)'H');
	mappedByteBuffer.put(3,(byte)'9');
	mappedByteBuffer.put(5,(byte)'Y');
	//IndexOutOfBoundsExceptionrandomAccessFile.close();System.out.println("修改成功~~");

三、 Selector

（1）Java的NIO，用非阻塞的IO方式。可以用一个线程，处理多个的客户端连接，就会使用到Selector(选择器)
（2）Selector能够检测多个注册的通道上是否有事件发生(注意:多个Channel以事件的方式可以注册到同一个Selector)，如果有事件发生，便获取事件然后针对每个事件进行相应的处理。这样就可以只用一个单线程去管理多个通道，也就是管理多个连接和请求。

（3）只有在连接/通道真正有读写事件发生时，才会进行读写，就大大地减少了系统开销，并且不必为每个连接都创建一个线程，不用去维护多个线程4)避免了多线程之间的上下文切换导致的开销。
总结：
（1）Netty的IO线程NioEventLoop聚合了Selector(选择器，也叫多路复用器)，可以同时并发处理成百上千个客户端连接。
（2）当线程从某客户端Socket通道进行读写数据时，若没有数据可用时，该线程可以进行其他任务。（3）线程通常将非阻塞IO的空闲时间用于在其他通道上执行IO操作，所以单独的线程可以管理多个输入和输出通道。
（4）由于读写操作都是非阻塞的，这就可以充分提升IO线程的运行效率，避免由于频繁I/O阻塞导致的线程挂起。
（5）一个I/O线程可以并发处理N个客户端连接和读写操作，这从根本上解决了传统同步阻塞I/O一连接一线程模型，架构的性能、弹性伸缩能力和可靠性都得到了极大的提升

四、 NIO非阻塞网络编程原理分析图

NIO非阻塞网络编程相关的(Selector、SelectionKey、ServerScoketChannel和SocketChannel)关系梳理图

（1）当客户端连接时，会通过ServerSocketChannel得到SocketChannel
（2）Selector进行监听select方法,返回有事件发生的通道的个数.
（3）将socketChannel注册到Selector上,register(Selectorsel,intops),一个selector上可以注册多个SocketChannel
（4）注册后返回一个SelectionKey,会和该Selector关联(集合)
（5）进一步得到各个SelectionKey(有事件发生)
（6）在通过SelectionKey反向获取SocketChannel,方法channel()
（7）可以通过得到的channel,完成业务处理

五、 NIO非阻塞网络编程

package xin.marico.facerecogition.test;

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;

public class NioServer {
    public static void main(String[] args) throws IOException {
        // 创建ServerSocketChannel->ServerSocket
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        //得到一个Selecor对象
        Selector selector = Selector.open();
        //绑定一个端口6666,在服务器端监听
        serverSocketChannel.socket().bind(new InetSocketAddress(6666));
        //设置为非阻塞
        serverSocketChannel.configureBlocking(false);

        //把serverSocketChannel注册到selector关心事件为OP_ACCEPT
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        //循环等待客户端连接
        while (true) {
            //这里我们等待1秒，如果没有事件发生,返回
            if (selector.select(1000) == 0) {
                //没有事件发生
                System.out.println("服务器等待了1秒，无连接");
                continue;
            }
            // 如果返回的>0,就获取到相关的selectionKey集合
            // 1.如果返回的>0，表示已经获取到关注的事件
            // 2.selector.selectedKeys()返回关注事件的集合
            // 通过selectionKeys反向获取通道
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            //遍历Set<SelectionKey>,使用迭代器遍历
            Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
            while (keyIterator.hasNext()) {
                //获取到SelectionKey
                SelectionKey key = keyIterator.next();
                //根据key对应的通道发生的事件做相应处理
                if (key.isAcceptable()) {
                    //如果是OP_ACCEPT,有新的客户端连接
                    // 该该客户端生成一个SocketChannel
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    System.out.println("客户端连接成功生成了一个socketChannel" + socketChannel.hashCode());
                    // 将SocketChannel设置为非阻塞
                    socketChannel.configureBlocking(false);
                    // 将socketChannel注册到selector,关注事件为OP_READ，同时给socketChannel
                    // 关联一个Buffer
                    socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
                }
                if (key.isReadable()) {//发生OP_READ
                    //通过key反向获取到对应channel
                    SocketChannel channel = (SocketChannel) key.channel();
                    //获取到该channel关联的buffer
                    ByteBuffer buffer = (ByteBuffer) key.attachment();
                    channel.read(buffer);
                    System.out.println("form客户端" + new String(buffer.array()));
                    buffer.clear();
                }
                //手动从集合中移动当前的selectionKey,防止重复操作
                keyIterator.remove();
            }
        }
    }
}

客户端.

package xin.marico.facerecogition.test;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

public class NioClient {
    public static void main(String[] args) throws IOException {
        // 得到一个网络通道
        SocketChannel socketChannel = SocketChannel.open();
        // 设置非阻塞
        socketChannel.configureBlocking(false);
        // 提供服务器端的ip和端口
        InetSocketAddress inetSoc ketAddress = new InetSocketAddress("127.0.0.1", 6666);
        // 连接服务器
        if (!socketChannel.connect(inetSocketAddress)) {
            while (!socketChannel.finishConnect()) {
                System.out.println("因为连接需要时间，客户端不会阻塞，可以做其它工作..");
            }
        }
        // 如果连接成功，就发送数据
        String str = "hello,NIO~";
        // 将String封装成
        ByteBuffer buffer = ByteBuffer.wrap(str.getBytes());
        // 发送数据，将buffer数据写入channel
        socketChannel.write(buffer);
        socketChannel.close();
    }
}

六、 SelectionKey

支持四种事件：读、写、接受请求、连接

 public static final int OP_READ = 1 << 0;
 public static final int OP_WRITE = 1 << 2;
 public static final int OP_CONNECT = 1 << 3;
 public static final int OP_ACCEPT = 1 << 4;

七、NIO网络编程应用实例-群聊系统

（1）编写一个NIO群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）
（2）实现多人群聊
（3）服务器端：可以监测用户上线，离线，并实现消息转发功能
（4）客户端：通过channel可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息(有服务器转发得到

1、Server端

package xin.marico.facerecogition.group;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;

public class GroupChatServer {

    private Selector selector;

    private ServerSocketChannel serverSocketChannel;

    private final int PORT = 6666;

    public GroupChatServer() {
        try {
            // 获取选择器
            selector = Selector.open();
            // 打开ServerSocketChannel
            serverSocketChannel = ServerSocketChannel.open();
            // 绑定端口
            serverSocketChannel.socket().bind(new InetSocketAddress(PORT));
            // 设置非阻塞模式
            serverSocketChannel.configureBlocking(false);
            // 将serverSocketChannel的接受请求事件注册在selector上
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("服务器已就绪...");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void readData(SelectionKey selectionKey) {
        SocketChannel socketChannel = null;
        // 得到channel
        socketChannel = (SocketChannel) selectionKey.channel();
        // 创建buffer
        ByteBuffer buffer = ByteBuffer.allocate(1024);

        try {
            int count = socketChannel.read(buffer);

            // 根据count的值做处理。
            if (count > 0) {
                // 将缓冲区的数据拿出来
                String mess = new String(buffer.array());
                System.out.println(socketChannel.getRemoteAddress() + "say :" + mess);
                // 将消息发送给别人
                sendMessage2Oters(socketChannel, mess);
            }
        } catch (IOException e) {
            e.printStackTrace();
            try {
                System.out.println(socketChannel.getRemoteAddress() + "离线了");
                socketChannel.close();
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }

    }

    private void sendMessage2Oters(SocketChannel socketChannel, String mess) {
        System.out.println("服务器转发消息中");
        // 遍历注册在selector上的所有key
        for (SelectionKey seletionKey : selector.keys()) {
            // 得到key对应的channel
            Channel channel = seletionKey.channel();
            if (channel instanceof SocketChannel && channel != socketChannel) {
                SocketChannel destChannel = (SocketChannel) channel;
                ByteBuffer buffer = ByteBuffer.wrap(mess.getBytes());
                /// 将 buffer写到dest
                try {
                    destChannel.write(buffer);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    public void listen() {
        while (true) {
            // 获取当前
            try {
                int count = selector.select();
                if (count > 0) {
                    // 得到SelectionKeys
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    // 迭代器遍历
                    while (iterator.hasNext()) {
                        // 遍历获取key
                        SelectionKey selectionKey = iterator.next();
                        // 如果是Accept事件
                        if (selectionKey.isAcceptable()) {
                            SocketChannel socketChannel = serverSocketChannel.accept();
                            socketChannel.configureBlocking(false);
                            // 将socketChannel注册到selector上
                            socketChannel.register(selector, SelectionKey.OP_READ);
                            System.out.println(socketChannel.getRemoteAddress() + "上线了...");
                        } else if (selectionKey.isReadable()) {
                            readData(selectionKey);
                        }
                        // 将当前key删除
                        iterator.remove();
                    }
                } else {
                    System.out.println("no message...");
                }
            } catch (IOException e) {
                e.printStackTrace();
                break;
            }

        }
    }

    public static void main(String[] args) {
        // 创建服务器对象
        GroupChatServer groupChatServer = new GroupChatServer();
        groupChatServer.listen();
    }
}

2、Client端

package xin.marico.facerecogition.group;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
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.Scanner;
import java.util.concurrent.TimeUnit;

public class GropChatClient {

    private Selector selector;

    private ServerSocketChannel serverSocketChannel;

    private final int PORT = 6666;

    private final String Host = "127.0.0.1";

    private String userName;

    private SocketChannel socketChannel;


    public GropChatClient() {
        // 获取选择器
        try {
            selector = Selector.open();
            // 连接服务器
            socketChannel = socketChannel.open(new InetSocketAddress(Host, PORT));
            // 设置非阻塞
            socketChannel.configureBlocking(false);
            // 将channel注册到selector
            socketChannel.register(selector, SelectionKey.OP_READ);
            //得到username
            userName = socketChannel.getLocalAddress().toString().substring(1);
            System.out.println(userName + "isok...");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //向服务器发送消息
    public void sendInfo(String info) {
        info = userName + "说：" + info;
        try {
            socketChannel.write(ByteBuffer.wrap(info.getBytes()));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //读取从服务器端回复的消息
    public void readInfo() {
        try {
            int readChannels = selector.select();
            if (readChannels > 0) {
                //有可以用的通道
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    if (key.isReadable()) {
                        //得到相关的通道3
                        SocketChannel sc = (SocketChannel) key.channel();
                        //得到一个Buffer
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        //读取
                        sc.read(buffer);
                        //把读到的缓冲区的数据转成字符串
                        String msg = new String(buffer.array());
                        System.out.println(msg.trim());
                    }
                }
                iterator.remove();
                //删除当前的selectionKey,防止重复操作
            } else {
                System.out.println("没有可以用的通道...");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        GropChatClient chatClient = new GropChatClient();
        // 启动一个线程，每隔三秒接受一下服务器的数据
        new Thread(() -> {
            while (true) {
                chatClient.readInfo();
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        // 获取键盘的输入
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            chatClient.sendInfo(scanner.nextLine());
        }
    }
}

总结

这里必须总结，要养成一个习惯，无论什么样的SocketChannel，都要注册到Selecor上，由Selecor进行维护。