1. Netty准备知识：Java NIO

2023-08-27 23:25:58

前言：我们知道，Netty是基于NIO开发的一套框架，在学习Netty之前，我们先学习下Java NIO。

一、IO多路复用模型

　　IO多路复用模型使用了Reactor设计模式，主要有三种实现：Reacotr单线程、Reactor多线程、Reactor主从模式。

1. Reactor单线程

　　在Reactor单线程模式中，所有客户端的请求处理都交给一个线程，串行化处理，效率较低。

2. Reactor多线程

　　在Reactor多线程模式中，acceptor线程负责接受客户端请求并将请求处理任务交给线程池，提升了请求处理速度。但是当client数量过多时，单线程就无法同时处理那么多的请求，造成瓶颈问题。

3. Reactor主从模式

　　为了解决Reactor多线程请求转发瓶颈问题，Reactor主从模式将acceptor设计为线程池，用以处理客户端请求。

二、NIO使用示例

1. NIO服务端通讯示例

public class TimeServer {
    public static void main(String[] args) throws IOException {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                // 采用默认值
            }
        }
        MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
        new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();
    }
}

public class MultiplexerTimeServer implements Runnable {
    private Selector selector;
    private ServerSocketChannel servChannel;
    private volatile boolean stop;//保证线程可见（volitile关键字）

    /**
     * 初始化多路复用器、绑定监听端口
     */
    public MultiplexerTimeServer(int port) {
        try {
            selector = Selector.open();//创建多路复用器
            servChannel = ServerSocketChannel.open();//打开ServerSocketChannel，用于监听客户端链接
            servChannel.configureBlocking(false);//设置非阻塞
            servChannel.socket().bind(new InetSocketAddress(port), 1024);//绑定端口
            servChannel.register(selector, SelectionKey.OP_ACCEPT);//注册监听（监听ACCEPT事件）
            System.out.println("The time server is start in port : " + port);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

    public void stop() {
        this.stop = true;
    }

    /**
     * 无限轮询准备就绪的key，并对其进行处理
     */
    @Override
    public void run() {
        while (!stop) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.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 (Throwable t) {
                t.printStackTrace();
            }
        }

        // 多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源
        if (selector != null)
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
    }

    private void handleInput(SelectionKey key) throws IOException {
        if (key.isValid()) {
            // 处理新接入的请求消息
            if (key.isAcceptable()) {
                // Accept the new connection
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                // Add the new connection to the selector
                sc.register(selector, SelectionKey.OP_READ);
            }
            if (key.isReadable()) {
                // Read the data
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println("The time server receive order : " + body);
                    String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)
                            ? new java.util.Date(System.currentTimeMillis()).toString()
                            : "BAD ORDER";
                    doWrite(sc, currentTime);
                } else if (readBytes < 0) {
                    // 对端链路关闭
                    key.cancel();
                    sc.close();
                } else {
                    // 读到0字节，忽略
                }
            }
        }
    }

    private void doWrite(SocketChannel channel, String response) throws IOException {
        if (response != null && response.trim().length() > 0) {
            byte[] bytes = response.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            channel.write(writeBuffer);
        }
    }
}

服务端通讯序列图如下：

2. 客户端通讯示例

public class TimeClient {
    public static void main(String[] args) {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                // 采用默认值
            }
        }
        new Thread(new TimeClientHandle("127.0.0.1", port), "TimeClient-001").start();
    }
}

public class TimeClientHandle implements Runnable {
    private String host;
    private int port;
    private Selector selector;
    private SocketChannel socketChannel;
    private volatile boolean stop;

    public TimeClientHandle(String host, int port) {
        this.host = host == null ? "127.0.0.1" : host;
        this.port = port;
        try {
            selector = Selector.open();// 创建多路复用器
            socketChannel = SocketChannel.open();// 打开SocketChannel
            socketChannel.configureBlocking(false);// 设置非阻塞模式
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

    /**
     * 创建连接，无线循环准备好的key并对其进行处理
     */
    @Override
    public void run() {
        try {
            // 创建连接
            doConnect();
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
        while (!stop) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.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);
            }
        }

        // 多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源
        if (selector != null) {
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }

    private void handleInput(SelectionKey key) throws IOException {
        if (key.isValid()) {
            // 判断是否连接成功
            SocketChannel sc = (SocketChannel) key.channel();
            if (key.isConnectable()) {
                if (sc.finishConnect()) {
                    sc.register(selector, SelectionKey.OP_READ);//注册READ事件
                    doWrite(sc);
                } else
                    System.exit(1);// 连接失败，进程退出
            }
            if (key.isReadable()) {
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println("Now is : " + body);
                    this.stop = true;
                } else if (readBytes < 0) {
                    // 对端链路关闭
                    key.cancel();
                    sc.close();
                } else {
                    // 读到0字节，忽略
                }
            }
        }
    }

    private void doConnect() throws IOException {
        // 如果直接连接成功，则注册到多路复用器上，发送请求消息，读应答
        if (socketChannel.connect(new InetSocketAddress(host, port))) {
            socketChannel.register(selector, SelectionKey.OP_READ);
            doWrite(socketChannel);
        } else
            socketChannel.register(selector, SelectionKey.OP_CONNECT);
    }

    private void doWrite(SocketChannel sc) throws IOException {
        byte[] req = "QUERY TIME ORDER".getBytes();
        ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
        writeBuffer.put(req);
        writeBuffer.flip();
        sc.write(writeBuffer);
        if (!writeBuffer.hasRemaining())
            System.out.println("Send order 2 server succeed.");
    }
}

客户端通讯序列图如下：

三、NIO类库介绍

1. 缓冲区Buffer

　　Buffer是一个对象，它包含一些要写入或读出的数据。在NIO中，所有的数据都是在缓冲区处理的，读数据时，直接读进缓冲区，写数据时，直接写在缓冲区。

　　Buffer类库继承关系：

我们最常用的就是ByteBuffer，这里介绍下ByteBuffer的几个参数及使用方法：

capacity	数组容量，创建后不可变
limit	当写数据到buffer中时，limit一般和capacity相等，当读数据时，limit代表buffer中有效数据的长度
position	位置，下一次要被读或被写的位置
mark	标记，调用mark()来设置mark=position，调用reset()可以让position恢复到标记位置
clear()	令position=0;limit=capacity;mark=-1; 但是不清除byte数组内容
reset()	把position设置成mark的值，相当于之前做过一个标记，现在要退回到之前标记的地方
flip()	令limit=position;position=0
allocate(int capavity)	从堆空间中分配一个容量大小为capacity的byte数组作为缓冲区的byte数据存储器
allocateDirect(int capacity)	在堆外内存中分配一个容量大小为capacity的byte数组作为缓冲区的byte数据存储器
wrap(byte[] array)	将byte数组包装成ByteBuffer

tips：ByteBuffer有两个比较重要的实现，HeapByteBuffer 和 DirectByteBuffer。

　　HeapByteBuffer：在堆内申请的内存，利于维护

　　DirectByteBuffer：在堆外申请的内存，实现零拷贝，提升数据操作速度（外部读取JVM堆中数据是先把JVM数据读到一个内存块中，然后在这个块里读取，使用堆外内存可省去这一步骤）。

2. 通道Channel

特性：
1. 既可以从通道中读取数据，又可以写数据到通道。但流的读写通常是单向的。 2. 通道可以异步地读写。 3. 通道中的数据总是要先读到一个Buffer，或者总是要从一个Buffer中写入。重要实现： FileChannel：从文件中读写数据 DatagramChannel：通过UDP读写网络中的数据，因为UDP是无连接的网络协议，所以不能像其它通道那样读取和写入，它发送和接收的是数据包 SocketChannel：通过TCP读写网络中的数据 ServerSocketChannel：监听新进来的TCP连接，对每一个连接创建一个SocketChannel

　　示例（这里只举例了FileChannel 和 DatagramChannel，其他两种看上面第二部分）：

//FileChannel示例：

RandomAccessFile aFile = new RandomAccessFile("data/nio-data.txt", "rw");

FileChannel inChannel = aFile.getChannel();
// 设置1MB的缓冲区

ByteBuffer buf = ByteBuffer.allocate(1024);
// 读取数据到buf中，并返回字节数

int bytesRead = inChannel.read(buf);

while (bytesRead != -1) {

    System.out.println("Read " + bytesRead);

    buf.flip(); // 重设缓冲区 postion = 0 ,limit = 原本position

    while(buf.hasRemaining()){ //缓冲区中是否还有内容
        // 或者使用buf.get(bytes),将数据读进字节数组中

        System.out.print((char) buf.get());

    }

    buf.clear();//清空缓存区
    // 缓冲区只有1MB大小，需要循环读取

    bytesRead = inChannel.read(buf);

}
aFile.close();



//DataGramChannel：

// 打开连接
DatagramChannel channel = DatagramChannel.open();
channel.socket().bind(new InetSocketAddress(port));

// buffer接收channel的数据
channel.receive(buf);//如果buffer容不下收到的数据，多出的数据将被抛弃

// 发送数据
channel.send(buf, new InetSocketAddress("baidu.com", 80));

// 连接到特定地址：由于UDP是无连接的，连接到特定的地址并不会像TCP通道那样创建一个真正的连接，而是锁住DatagramChannel，让其只能从特定地址收发数据
channel.connect(new InetSocketAddress("baidu.com", 80));
连接后，可以使用read()和write()方法，但数据传送无保证
channel.read(buf);
channel.write(buf);

Channel示例

3. 多路复用器Selector

　　Selector不断轮询注册在其上的Channel，如果某个Channel上有新的TCP连接、读、写操作，这个Channel就处于就绪状态，会被Selector轮询出来，然后通过SelectionKey就可以获取到就绪Channel集合，进行后续的IO操作。

　　一些方法：

// 1. 创建Selector
Selector selector = Selector.open();

// 2. 注册通道
channel.configureBlocking(false);//channel必须处于非阻塞状态
SelectionKey key = channel.register(selector, Selector.OP_READ);
   
int select()：阻塞到至少有一个通道在你注册的事件上就绪了，返回值为多少通道已就绪
int select(long timeout)：同上，最长阻塞timeout毫秒
int selectNow()：不阻塞，不管什么通道都立即返回
Set selectedKeys = selector.selectedKeys()：返回已就绪的通道的SelectedKey

我们获取到的是SelectionKey，该对象中包含了一些有价值的属性：

insterest集合：OP_CONNECT、OP_ACCEPT、OP_READ、OP_WRITE   
    int interestSet = selectionKey.interestOps();
    boolean isInterestedInAccept  = interestSet & SelectionKey.OP_ACCEPT；
    boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
    boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;
    boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;

ready集合：已准备就绪的操作的集合（你注册了监听connect，那么除了connect其它都是false），在一次选择(Selection)之后，会首先访问这个ready set
     int readySet = selectionKey.readyOps();
     boolean isAccept = selectionKey.isAcceptable();
     boolean isConnect = selectionKey.isConnectable();
     boolean isReadable = selectionKey.isReadable();
     boolean isWritable = selectionKey.isWritable();

channel：Channel  channel  = selectionKey.channel();
selector：Selector selector = selectionKey.selector();
附加对象(可选)：SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

码农公寓