Netty源码学习系列①JavaNIO概览

不念不忘少年蓝@ 2023-07-06 13:04 29阅读 0赞

> 在正式开始Netty相关的学习之前，我决定还是要先回顾一下Java NIO，至少要对Java NIO相关的概念有一个了解，如Channel、ByteBuffer、Selector等。要自己动手写一写相关的demo实例、并且要尽可能地去了解其后面是如何实现的，也就是稍微看看相关jdk的源代码。

Java NIO 由以下几个核心部分组成：Buffer, Channel, Selector。传统的IO操作面向数据流，面向流 的 I/O 系统一次一个字节地处理数据，意味着每次从流中读一个或多个字节，直至完成，数据没有被缓存在任何地方；NIO操作面向缓冲区（ 面向块），数据从Channel读取到Buffer缓冲区，随后在Buffer中处理数据。

## Buffer ##

可以理解成煤矿里面挖煤的小车，把煤从井底运地面上面。它的属性与子类如下：

![format_png][]

Buffer是一个抽象类，继承自Object，拥有多个子类。此类在JDK源码中的注释如下：

> A container for data of a specific primitive type.
> 
> A buffer is a **linear, finite sequence of elements of a specific primitive type**. Aside from its content, the essential properties of a buffer are its `capacity`, `limit`, and `position`:
> 
>  *  A buffer’s `capacity` is **the number of elements it contains**. The capacity of a buffer is never negative and never changes.
>  *  A buffer’s `limit` is **the index of the first element that should not be read or written**. A buffer’s limit is never negative and is never greater than its capacity.
>     
>     **写模式下**，limit表示最多能往Buffer里写多少数据，等于capacity值；**读模式下**，limit表示最多可以读取多少数据，小于等于 capacity 值。
>  *  A buffer’s `position` is **the index of the next element to be read or written**. A buffer’s position is never negative and is never greater than its limit.
> 
> There is one subclass of this class for each non-boolean primitive type.
> 
> ## Transferring data ##
> 
> Each subclass of this class defines **two categories of get and put operations**:
> 
>  *  Relative operations read or write one or more elements starting at the current position and then increment the position by the number of elements transferred. If the requested transfer exceeds the limit then a relative get operation throws a `BufferUnderflowException` and a relative put operation throws a `BufferOverflowException`; in either case, no data is transferred.
>  *  Absolute operations take an explicit element index and do not affect the position. Absolute get and put operations throw an `IndexOutOfBoundsException` if the index argument exceeds the limit.
> 
> Data may also, of course, be transferred in to or out of a buffer by the I/O operations of an appropriate channel, which are always relative to the current position.
> 
> ## Marking and resetting ##
> 
> A buffer’s `mark` is **the index to which its position will be reset when the reset method is invoked**. The mark is not always defined, but when it is defined it is never negative and is never greater than the position. If the mark is defined then it is discarded when the position or the limit is adjusted to a value smaller than the mark. If the mark is not defined then invoking the reset method causes an `InvalidMarkException` to be thrown.
> 
> ## Invariants ##
> 
> The following invariant holds for the mark, position, limit, and capacity values:
> 
> **`0 <= mark <= position <= limit <= capacity`**
> 
> A newly-created buffer always has a position of zero and a mark that is undefined. The initial limit may be zero, or it may be some other value that depends upon the type of the buffer and the manner in which it is constructed. Each element of a newly-allocated buffer is initialized to zero.
> 
> ## Clearing, flipping, and rewinding ##
> 
> In addition to methods for accessing the `position`, `limit`, and `capacity` values and for marking and resetting, this class also defines the following operations upon buffers:
> 
>  *  `clear()` **makes a buffer ready for a new sequence of channel-read or relative put operations**: It sets the limit to the capacity and the position to zero.
>  *  `flip()` **makes a buffer ready for a new sequence of channel-write or relative get operations**: It sets the limit to the current position and then sets the position to zero.
>  *  `rewind()` **makes a buffer ready for re-reading the data that it already contains**: It leaves the limit unchanged and sets the position to zero.
> 
> ## Read-only buffers ##
> 
> Every buffer is readable, but not every buffer is writable. The mutation methods of each buffer class are specified as optional operations that will throw a `ReadOnlyBufferException` when invoked upon a read-only buffer. A read-only buffer does not allow its content to be changed, but its mark, position, and limit values are mutable. Whether or not a buffer is read-only may be determined by invoking its isReadOnly method.
> 
> ## Thread safety ##
> 
> Buffers are not safe for use by multiple concurrent threads. If a buffer is to be used by more than one thread then access to the buffer should be controlled by appropriate synchronization.
> 
> ## Invocation chaining ##
> 
> Methods in this class that do not otherwise have a value to return are specified to return the buffer upon which they are invoked. This allows method invocations to be chained; for example, the sequence of statements
> 
>     b.flip();
>     b.position(23);
>     b.limit(42);
> 
> can be replaced by the single, more compact statement  
> `b.flip().position(23).limit(42);`

clear()方法

// 清除Buffer中的信息，只将参数恢复成默认
    public final Buffer clear() {
        
        position = 0;
        limit = capacity;
        mark = -1;
        return this;
    }

flip()方法

// 将limit记录成当前的位置，指针指向头部，为读取做准备
    public final Buffer flip() {
        
        limit = position;
        position = 0;
        mark = -1;
        return this;
    }

rewind()方法

// 指针指向头部，可以用于再次读取
    public final Buffer rewind() {
        
        position = 0;
        mark = -1;
        return this;
    }

## 如何使用Java NIO读取文件内容 ##

遇到坑了，但是感觉可以透过这个问题更加深入理解Java NIO的这些概念。出现问题的代码：

public static void fileChannel() throws IOException {
        
        FileInputStream fis = new FileInputStream("/Users/yangyu/Documents/data.json");
        FileChannel fileChannel = fis.getChannel();
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        while ((fileChannel.read(byteBuffer)) != -1) {
        
            while (byteBuffer.hasRemaining()) {
        
                System.out.print((char) byteBuffer.get());
            }
        }
    }

上面的代码读取不到数据，一直在做循环，但是不输出数据。为什么？因为`hasRemaining()`是以`position`和`limit`作对比，如下：

public final boolean hasRemaining() {
        
    		return position < limit;
    }

当从`fileChannel`中读取数据到`byteBuffer`中之后，`limit`与`capacity`相等（初始化既如此），此时的`position`也与`capacity`相同，导致`hasRemaining()`为`false`，无法向控制台输出。

![format_png 1][]

所以需要将position设置成从0开始，让读取从0开始，直到读到之前的容量，所以使用`flip()`来完成这个目的，即：

![format_png 2][]

此时却发现控制台无限打印东西，为了弄明白这是为什么，我把`byteBuffer`的大小调成了8，跑起来之后的输出如下：

![format_png 3][]

这是为什么呢？这个问题应该与`byteBuffer`里面的那几个参数有关系：

![format_png 4][]

猜测应该是与`fileChannel.read(byteBuffer)`中的具体实现有关。粗略看了看`fileChannel.read(byteBuffer)`的实现，大致流程如下：

1.  计算`byteBuffer`的剩余量，即`limit - position`。对于上面的情况，剩余量为0。
2.  找出缓存buffer，此时缓存buffer为上次read得到的，第一次为空会直接分配；第二次`read`的时候，其3大属性全部为8，也即上次读取的结果。
3.  将缓存的buffer进行`rewind()`、`flip(剩余量)`，得到一个\[pos=0, limit=0, capacity=8\]的buffer。
4.  进行读取的时候回根据缓存buffer的pos、limit来确定能读取的数量，也即：
    
        // 其中var1为缓存buffer
        int var5 = var1.position();
        int var6 = var1.limit();
        
        assert var5 <= var6;
        // var6 - var5 = limit - position = 0
        int var7 = var5 <= var6 ? var6 - var5 : 0;
        // var7 = 0
        if (var7 == 0) {
              
          // 0 即读取的字节数
          return 0;
        }
5.  如果能读到数据，会将缓存buffer里面的的内容再转移到`byteBuffer`（也就是我们`read()`里面传的`ByteBuffer`）中：
    
        // var5即缓存buffer，读取内容到var5中
        int var6 = readIntoNativeBuffer(var0, var5, var2, var4);
        // 准备用来读取
        var5.flip();
        // var1是我们传入的byteBuffer，如果读取到的字节数大于0，
        if (var6 > 0) {
              
          // 将var5中的内容拷贝到var1中
          var1.put(var5);
        }

直到发现`flip()`的注释里面有这样一段注释：

> Compacts this buffer (optional operation).  
> The bytes between the buffer’s current position and its limit, if any, are copied to the beginning of the buffer. That is, the byte at index `p = position()` is copied to index zero, the byte at index `p + 1` is copied to index one, and so forth until the byte at index `limit() - 1` is copied to index `n = limit() - 1 - p`. The buffer’s position is then set to `n+1` and its limit is set to its `capacity`. The `mark`, if defined, is discarded.
> 
> The buffer’s `position` is set to the number of bytes copied, rather than to zero, so that an invocation of this method can be followed immediately by an invocation of another relative put method.
> 
> Invoke this method after writing data from a buffer in case the write was incomplete. The following loop, for example, copies bytes from one channel to another via the buffer buf:
> 
>     buf.clear();          // Prepare buffer for use
>     while (in.read(buf) >= 0 || buf.position != 0) {
>          
>       buf.flip();
>       out.write(buf);
>       buf.compact();    // In case of partial write
>     }

加上这段代码`buf.compact()`便可以正常读取文件内容。到这里就有点心累了，为什么写个读取都这么多坑。感觉有问题的时候往这三个参数上面想就行了。

## Channel ##

煤矿厂里面运煤的通道，需要看的子类总共有4个，分别为：

*  `FileChannel`：文件通道，用于文件的读和写。不支持非阻塞
 *  `DatagramChannel`：用于 UDP 连接的接收和发送
 *  `SocketChannel`：把它理解为 TCP 连接通道，简单理解就是 TCP 客户端
 *  `ServerSocketChannel`：TCP 对应的服务端，用于监听某个端口进来的请求

![format_png 5][]

## Selector ##

只有自己写过的代码才会有更深刻的印象，哪怕是从别的地方抄来的，自己慢慢debug一下，找出自己对代码的疑问，然后再去搞清楚这些问题，我觉得这样让我对它的了解会更深。这两段代码基本和网上的教程类似，大部分是抄的，但是自己有一定的加工，也遇到了1个问题，外加一个疑问。

### 客户端代码 ###

客户端的代码很简单：①读标准输入。②发送给Server端。

// Client端的代码很像八股文，这样弄就行了。
    public static void main(String[] args) throws Exception {
        
        SocketChannel sc = SocketChannel.open();
        sc.configureBlocking(false);
        sc.connect(new InetSocketAddress("127.0.0.1", 8086));
    
        Scanner scanner = new Scanner(System.in);
        if (sc.finishConnect()) {
        
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            while (scanner.hasNextLine()) {
        
              	// 读标准输入
                String info = scanner.nextLine();
                buffer.clear();
                buffer.put(info.getBytes());
                buffer.flip();
                while (buffer.hasRemaining()) {
        
                    System.out.println(buffer);
                  	// 发送给Server端
                    sc.write(buffer);
                }
            }
        }
    }

### 服务端代码 ###

主要参考了[一篇CSDN上的博客][CSDN]和[一篇简书上的博客][Link 1]，简书上面的这边对我的帮助很大，十分感谢。我的问题主要有两点，第一个是少了`it.remove();`，第二个是关于如何触发`SelectionKey.OP_WRITE`事件。

// 八股文的感觉。
    public static void startServer() throws IOException {
        
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        ssc.socket().bind(new InetSocketAddress(8086));
    
        Selector selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT);
    
        while (true) {
        
            selector.select();
            Iterator<SelectionKey> it = selector.selectedKeys().iterator();
            while (it.hasNext()) {
        
                SelectionKey key = it.next();
                // 一定要remove掉，不然上次的事件会累积。
                // 也就是对同一事件会处理两次，这样可能会导致报错。
                it.remove();
                if (key.isAcceptable()) {
        
                    System.out.println("ACCEPT");
                    ServerSocketChannel ssChannel = (ServerSocketChannel)key.channel();
                    SocketChannel sc = ssChannel.accept();
                    sc.configureBlocking(false);
                    sc.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocateDirect(BUF_SIZE));
                } else if (key.isReadable()) {
        
                    System.out.print("READ：");
                    SocketChannel sc = (SocketChannel)key.channel();
                    ByteBuffer buf = (ByteBuffer)key.attachment();
                    long bytesRead = sc.read(buf);
                    while(bytesRead>0){
        
                        buf.flip();
                        while(buf.hasRemaining()){
        
                            System.out.print((char)buf.get());
                        }
                        System.out.println();
                        buf.clear();
                        bytesRead = sc.read(buf);
                    }
                    if(bytesRead == -1){
        
                        sc.close();
                    }
                } else if (key.isWritable()) {
        
                    // OP_WRITE事件如何触发？
                    System.out.print("WRITE：");
                    ByteBuffer buf = (ByteBuffer) key.attachment();
                    buf.flip();
                    SocketChannel sc = (SocketChannel) key.channel();
                    while(buf.hasRemaining()){
        
                        sc.write(buf);
                    }
                    buf.compact();
                } else if (key.isConnectable()) {
        
                    System.out.println("CONNECT");
                } else {
        
                    System.out.println("UNKNOWN");
                }
            }
        }
    }

1.  如果缺少`it.remove()`方法的调用，那么会导致事件会堆积在Selector的`Set<SelectionKey> publicSelectedKeys`中，引发对同一事件会处理两次，这样可能会导致报错。
2.  如何触发`SelectionKey.OP_WRITE`？因为我看到大部分关于selector的博客，都没有写如何触发该事件，并且也未对读事件做出说明。

首先肯定要在调用`ssChannel.accept()`之后，将得到的`SocketChannel`多注册一个`OP_WRITE`事件。即修改成：

SocketChannel sc = ssChannel.accept();
    sc.configureBlocking(false);
    sc.register(key.selector(), SelectionKey.OP_READ | SelectionKey.OP_WRITE,ByteBuffer.allocateDirect(BUF_SIZE));

然后会发现程序卡死，屏幕一直输出`Write`。为什么会有这么多`OP_WRITE`事件？因为Java NIO的事件触发是**水平触发**，即**只要满足条件，就触发一个事件**，所以**只要内核缓冲区还不满，就一直发出`OP_WRITE`事件**。

与水平触发对应的还有一个叫做**边缘触发**，即**每当状态变化时，触发一个事件**。对之前的Netty的事件是边缘触发又有了一个认识。

![format_png 6][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjEyMjA0MDUwLnBuZw?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjEyMjIxNjA4LnBuZw?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjE2MjI1NTEyLnBuZw?x-oss-process=image/format,png
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjE2MjI1ODE5LnBuZw?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjE2MjMxMTU0LnBuZw?x-oss-process=image/format,png
[format_png 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjE3MjMzMTE5LnBuZw?x-oss-process=image/format,png
[CSDN]: https://blog.csdn.net/u011381576/article/details/79876754
[Link 1]: https://www.jianshu.com/p/6bdee8cfee90
[format_png 6]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjE4MjIwMjI4LnBuZw?x-oss-process=image/format,png