Netty全套课程

使用Netty开发基本网络应用程序
彻底理解阻塞、非阻塞的区别，并跟Netty、NIO的编码联系起来
懂得多路复用在服务器开发时的优势，为什么在此基础上还要加多线程
Netty 中是如何实现异步的，异步处理的优势是什么
Netty中是如何管理线程的，EventLoop 如何运作
Netty中是如何管理内存的，ByteBuf特点和分配时机
看源码、调试的一些技巧，让自己也能去看、去跟源码

maven官网

Maven Repository: Search/Browse/Explore (mvnrepository.com)

NIO之bytebuffer内部结构和方法 - 死磕 Java (skjava.com)

Netty - Nyima's Blog (gitee.io)

Netty学习之NIO基础 - Nyima's Blog (gitee.io)

netty系列之:NIO和netty详解 - 知乎 (zhihu.com)

基础

1. 三大组件

netty为什么快呢？这是因为netty底层使用了JAVA的NIO技术，并在其基础上进行了性能的优化，虽然netty不是单纯的JAVA nio，但是netty的底层还是基于的是nio技术。

nio是JDK1.4中引入的，用于区别于传统的IO，所以nio也可以称之为new io。

nio的三大核心是Selector,channel和Buffer

Channel

channel有一点类似于 stream,它就是读写数据的双向通道，可以从channel将数据读入buffer，也可以将buffer的数据写入channel.而之前的stream要么就是输入，要么就是输出。channel比stream更为底层。

mermaid

graph LR
channel(channel)-->F[buffer]
F-->channel

常见的Channel有

FileChannel
DatagramChannel
SocketChannel
ServerSocketChannel

Buffer

buffer则用来缓冲读写数据，.

ByteBuffer
- MappedByteBuffer
- DirectByteBuffer
- HeapByteBuffer
ShortBuffer
lntBuffer
LongBuffer
FloatBuffer
DoubleBuffer
CharBuffer

Selector

selector单从字面意思不好理解，需要结合服务器的设计演化来理解它的用途

多线程设计模板

mermaid

graph TB
thread(thread)-->socket1
thread1(thread)-->socket2
thread2(thread)-->socket3

多线程版缺点

内存占用高
线程上下文切换成本高
只适合连接数少的场景

线程池版设计

mermaid

graph TB
thread(thread)-->socket1
thread-->socket2
thread1(thread)-->socket3
thread1(thread)-->socket4

线程池版缺点

阻塞模式下，线程仅能处理一个socket连接
仅适合短连接场景

selector版设计

selector的作用就是配合一个线程来管理多个channel，获取这些channel上发生的事件，这些channel工作在非阻塞模式下，不会让线程吊死在一个channel上。适合连接数特别多，但流星低的场景（low traffic)

mermaid

graph TB
thread(thread)-->selector
selector-->channel1(channel)
selector-->socket2(channel)
selector-->socket3(channel)

调用selector的 select()会阻塞直到channel发生了读写就绪事件，这些事件发生，select方法就会返回这些事件交给thread来处理。

2. ByteBuffer

1) 入门Demo

首先准备一个Maven项目，导入pom文件

xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>mynetty</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>

        <!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.100.Final</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/cn.hutool/hutool-all -->
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.8.22</version>
        </dependency>


        <!-- https://mvnrepository.com/artifact/com.google.code.gson/gson -->
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.10.1</version>
        </dependency>


        <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.30</version>
            <scope>provided</scope>
        </dependency>


    </dependencies>

</project>

然后准备一个data.txt

txt

0123456789abc
2343
hi

编写一个bytebuffer测试文件

java

package cn.itcast.netty.c1;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

public class FileTest {

    public static void main(String[] args) {

        // FileChannel
        // 1. 输入输出流 2. RandomAccessFile

        try (FileChannel fileChannel = new FileInputStream("data.txt").getChannel()) {
            // 准备10字节作为我们的缓冲区
            ByteBuffer buffer = ByteBuffer.allocate(10);
            // 从channel 读取数据，向buffer写入
            while (true) {
                int len = fileChannel.read(buffer); // 当返回-1时。表示当前流已经到末尾了。接下来没有数据了
                System.out.printf("读取到了%d 字节\n",len);
                if (len < 0) {
                    break;
                }
                // 打印buffer的内容
                buffer.flip(); // 切换至读模式
                while (buffer.hasRemaining()) { // 是否还有剩余未读数据
                    byte b = buffer.get(); // 每次能够读取 10字节
                    System.out.println((char) b);
                }
                buffer.clear(); // 切换至写模式
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

ByteBuffer正确使用姿势

向buffer 写入数据，例如调用channel.read(buffer)
调用flip()切换至读模式
从buffer 读取数据，例如调用buffer.get()
调用clear()或compact()切换至写模式
重复1~4步骤

2) ByteBuffer内部结构

看一下底层

java

public abstract class ByteBuffer
    extends Buffer
    implements Comparable<ByteBuffer>
{
    // Cached array base offset
    private static final long ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);

    // These fields are declared here rather than in Heap-X-Buffer in order to
    // reduce the number of virtual method invocations needed to access these
    // values, which is especially costly when coding small buffers.
    //
    final byte[] hb;                  // Non-null only for heap buffers
    final int offset;
    boolean isReadOnly;
}

Buffer底层

java

public abstract class Buffer {
    // Cached unsafe-access object
    static final Unsafe UNSAFE = Unsafe.getUnsafe();

    static final ScopedMemoryAccess SCOPED_MEMORY_ACCESS = ScopedMemoryAccess.getScopedMemoryAccess();

    /**
     * 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;

    // Used by heap byte buffers or direct buffers with Unsafe access
    // For heap byte buffers this field will be the address relative to the
    // array base address and offset into that array. The address might
    // not align on a word boundary for slices, nor align at a long word
    // (8 byte) boundary for byte[] allocations on 32-bit systems.
    // For direct buffers it is the start address of the memory region. The
    // address might not align on a word boundary for slices, nor when created
    // using JNI, see NewDirectByteBuffer(void*, long).
    // Should ideally be declared final
    // NOTE: hoisted here for speed in JNI GetDirectBufferAddress
    long address;

    // Used by buffers generated by the memory access API (JEP-370)
    final MemorySegmentProxy segment;
}

属性介绍

Bytebuffer有以下重要属性：

capacity（容量） ：缓冲区的容量。通过构造函数赋予，一旦设置，无法更改。
position（指针） ：读写指针，记录数据读写的位置，缓冲区的位置不能为负，并且 不能大于limit 。
limit（读写限制） ：缓冲区的界限。位于limit 后的数据不可读写。缓冲区的限制不能为负，并且 不能大于其容量 。
mark：记录当前position的值。position被改变后，可以通过调用reset() 方法恢复到mark的位置。

以上四个属性必须满足以下要求

mark <= position <= limit <= capacity

结构

写模式

当缓冲区刚创建成功时

写模式下，position 是写入位置，limit 等于容量。

下图表示写入了 4 个字节后的状态：

Position移动到第5个字节开始位置

读模式

flip动作发生后，position切换为读取位置，limit切换为读取限制

position重新赋值到开始位置，因为读取数据从开始位置开始读取
limit被赋值为position写入时的最后位置，作为数据读取的最终位置

读取4个字节后，状态如下图：

当position位置与limit位置相同时，数据读取结束。
数据未读取完重新切换到写模式时

compact 方法，是把未读完的部分向前压缩，然后切换至写模式

常见方法

allocate()分配内存空间
可以使用allocate() 和 allocateDirect()方法为ByteBuffer分配空间，其他buffer类也有该方法
- allocate() : 使用的是java的堆内存，堆内字节缓冲区，读写效率低，会受到GC的影响
- allocateDirect() :使用的是直接内存，直接内存字节缓冲区，读写效率高（零拷贝），不会受GC影响,因为是系统直接内存，所以分配内存要调用操作系统函数，所以分配内存的速度较慢，如果使用不当（资源没得到合理释放），会造成内存泄漏。
java
```
 import java.nio.ByteBuffer;
    
    /**
     * Created by lilinchao
     * Date 2022/5/25
     * Description 分配空间
     */
    public class ByteBufferAllocateDemo {
        public static void main(String[] args) {
            System.out.println(ByteBuffer.allocate(16).getClass());
            System.out.println(ByteBuffer.allocateDirect(16).getClass());
        }
    }
```

put()
描述：可以将一个数据放入到缓冲区中。
进行该操作后，postition的值会+1，指向下一个可以放入的位置。capacity = limit ，为缓冲区容量的值。
flip()
描述：用来 切换对缓冲区的操作模式 ，由写->读 / 读->写
进行该操作后：
- 如果是写模式->读模式，position = 0 ， limit 指向最后一个元素的下一个位置，capacity不变；
- 如果是读模式->写模式，则恢复为put()方法中的值。
get()
描述：该方法会读取缓冲区中的一个值
进行该操作后，position会+1，如果超过了limit则会抛出异常
注意：get(i)方法不会改变position的值

rewind()

描述：该方法 只能在读模式下使用

rewind()方法后，会恢复position、limit和capacity的值，变为进行get()前的值

java

package cn.itcast.netty.c1;

import java.nio.ByteBuffer;

public class TestByteBufferRead {

    public static void main(String[] args) {
        ByteBuffer buffer = ByteBuffer.allocate(10);
        buffer.put(new byte[]{'a', 'b', 'c', 'd'});


        buffer.flip();
        buffer.get(new byte[4]); // 一次性获取全部的值

        buffer.rewind(); // 将position的值变成0.从头开始读取
        /**
         *   public Buffer rewind() {
         *         position = 0;
         *         mark = -1;
         *         return this;
         *     }
         */
        System.out.println((char) buffer.get());
        
        
    }
}

clean()
描述：会将缓冲区中的各个属性恢复为最初的状态，position = 0, capacity = limit
此时缓冲区的数据依然存在 ，处于“被遗忘”状态，下次进行写操作时会覆盖这些数据

mark()/reset()

描述：mark 是在读取时，做一个标记，即使 position 改变，只要调用 reset 就能回到 mark 的位置

mark() ：将postion的值保存到mark属性中

reset() ：将position的值改为mark中保存的值

注意：rewind 和 flip 都会清除 mark 位置

java

package cn.itcast.netty.c1;

import java.nio.ByteBuffer;

public class TestByteBufferRead {

    public static void main(String[] args) {
        ByteBuffer buffer = ByteBuffer.allocate(10);
        buffer.put(new byte[]{'a', 'b', 'c', 'd'});


        buffer.flip();

        System.out.println((char) buffer.get()); // get结束后  position =1
        System.out.println((char) buffer.get()); // get结束后 position =2

        buffer.mark(); // position =2 标记第3个位置
        System.out.println((char) buffer.get()); // get结束后 position =3
        System.out.println((char) buffer.get());// get结束后 position =4

        buffer.reset(); // position 从4 变成 2
        System.out.println((char) buffer.get());

    }
}

compact()
描述：compact会把未读完的数据向前压缩，然后切换到写模式
数据前移后，原位置的值并未清零，写时会覆盖之前的值
注意：此方法为ByteBuffer的方法，而不是Buffer的方法
clear()和compact()方法对比：
clear只是对position、limit、mark进行重置，而compact在对position进行设置，以及limit、mark进行重置的同时，还涉及到数据在内存中拷贝（会调用arraycopy）。所以compact比clear更耗性能。但compact能保存你未读取的数据，将新数据追加到为读取的数据之后；而clear则不行，若你调用了clear，则未读取的数据就无法再读取到了

3) 方法调用及演示

调试工具类(重要)

java

package cn.itcast.netty.c1;

import java.nio.ByteBuffer;

import io.netty.util.internal.MathUtil;
import io.netty.util.internal.StringUtil;
import io.netty.util.internal.MathUtil.*;


/**
 * @author Panwen Chen
 * @date 2021/4/12 15:59
 */
public class ByteBufferUtil {
    private static final char[] BYTE2CHAR = new char[256];
    private static final char[] HEXDUMP_TABLE = new char[256 * 4];
    private static final String[] HEXPADDING = new String[16];
    private static final String[] HEXDUMP_ROWPREFIXES = new String[65536 >>> 4];
    private static final String[] BYTE2HEX = new String[256];
    private static final String[] BYTEPADDING = new String[16];

    static {
        final char[] DIGITS = "0123456789abcdef".toCharArray();
        for (int i = 0; i < 256; i++) {
            HEXDUMP_TABLE[i << 1] = DIGITS[i >>> 4 & 0x0F];
            HEXDUMP_TABLE[(i << 1) + 1] = DIGITS[i & 0x0F];
        }

        int i;

        // Generate the lookup table for hex dump paddings
        for (i = 0; i < HEXPADDING.length; i++) {
            int padding = HEXPADDING.length - i;
            StringBuilder buf = new StringBuilder(padding * 3);
            for (int j = 0; j < padding; j++) {
                buf.append("   ");
            }
            HEXPADDING[i] = buf.toString();
        }

        // Generate the lookup table for the start-offset header in each row (up to 64KiB).
        for (i = 0; i < HEXDUMP_ROWPREFIXES.length; i++) {
            StringBuilder buf = new StringBuilder(12);
            buf.append(StringUtil.NEWLINE);
            buf.append(Long.toHexString(i << 4 & 0xFFFFFFFFL | 0x100000000L));
            buf.setCharAt(buf.length() - 9, '|');
            buf.append('|');
            HEXDUMP_ROWPREFIXES[i] = buf.toString();
        }

        // Generate the lookup table for byte-to-hex-dump conversion
        for (i = 0; i < BYTE2HEX.length; i++) {
            BYTE2HEX[i] = ' ' + StringUtil.byteToHexStringPadded(i);
        }

        // Generate the lookup table for byte dump paddings
        for (i = 0; i < BYTEPADDING.length; i++) {
            int padding = BYTEPADDING.length - i;
            StringBuilder buf = new StringBuilder(padding);
            for (int j = 0; j < padding; j++) {
                buf.append(' ');
            }
            BYTEPADDING[i] = buf.toString();
        }

        // Generate the lookup table for byte-to-char conversion
        for (i = 0; i < BYTE2CHAR.length; i++) {
            if (i <= 0x1f || i >= 0x7f) {
                BYTE2CHAR[i] = '.';
            } else {
                BYTE2CHAR[i] = (char) i;
            }
        }
    }

    /**
     * 打印所有内容
     * @param buffer
     */
    public static void debugAll(ByteBuffer buffer) {
        int oldlimit = buffer.limit();
        buffer.limit(buffer.capacity());
        StringBuilder origin = new StringBuilder(256);
        appendPrettyHexDump(origin, buffer, 0, buffer.capacity());
        System.out.println("+--------+-------------------- all ------------------------+----------------+");
        System.out.printf("position: [%d], limit: [%d]\n", buffer.position(), oldlimit);
        System.out.println(origin);
        buffer.limit(oldlimit);
    }

    /**
     * 打印可读取内容
     * @param buffer
     */
    public static void debugRead(ByteBuffer buffer) {
        StringBuilder builder = new StringBuilder(256);
        appendPrettyHexDump(builder, buffer, buffer.position(), buffer.limit() - buffer.position());
        System.out.println("+--------+-------------------- read -----------------------+----------------+");
        System.out.printf("position: [%d], limit: [%d]\n", buffer.position(), buffer.limit());
        System.out.println(builder);
    }

    private static void appendPrettyHexDump(StringBuilder dump, ByteBuffer buf, int offset, int length) {
        if (MathUtil.isOutOfBounds(offset, length, buf.capacity())) {
            throw new IndexOutOfBoundsException(
                    "expected: " + "0 <= offset(" + offset + ") <= offset + length(" + length
                            + ") <= " + "buf.capacity(" + buf.capacity() + ')');
        }
        if (length == 0) {
            return;
        }
        dump.append(
                "         +-------------------------------------------------+" +
                        StringUtil.NEWLINE + "         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |" +
                        StringUtil.NEWLINE + "+--------+-------------------------------------------------+----------------+");

        final int startIndex = offset;
        final int fullRows = length >>> 4;
        final int remainder = length & 0xF;

        // Dump the rows which have 16 bytes.
        for (int row = 0; row < fullRows; row++) {
            int rowStartIndex = (row << 4) + startIndex;

            // Per-row prefix.
            appendHexDumpRowPrefix(dump, row, rowStartIndex);

            // Hex dump
            int rowEndIndex = rowStartIndex + 16;
            for (int j = rowStartIndex; j < rowEndIndex; j++) {
                dump.append(BYTE2HEX[getUnsignedByte(buf, j)]);
            }
            dump.append(" |");

            // ASCII dump
            for (int j = rowStartIndex; j < rowEndIndex; j++) {
                dump.append(BYTE2CHAR[getUnsignedByte(buf, j)]);
            }
            dump.append('|');
        }

        // Dump the last row which has less than 16 bytes.
        if (remainder != 0) {
            int rowStartIndex = (fullRows << 4) + startIndex;
            appendHexDumpRowPrefix(dump, fullRows, rowStartIndex);

            // Hex dump
            int rowEndIndex = rowStartIndex + remainder;
            for (int j = rowStartIndex; j < rowEndIndex; j++) {
                dump.append(BYTE2HEX[getUnsignedByte(buf, j)]);
            }
            dump.append(HEXPADDING[remainder]);
            dump.append(" |");

            // Ascii dump
            for (int j = rowStartIndex; j < rowEndIndex; j++) {
                dump.append(BYTE2CHAR[getUnsignedByte(buf, j)]);
            }
            dump.append(BYTEPADDING[remainder]);
            dump.append('|');
        }

        dump.append(StringUtil.NEWLINE +
                "+--------+-------------------------------------------------+----------------+");
    }

    private static void appendHexDumpRowPrefix(StringBuilder dump, int row, int rowStartIndex) {
        if (row < HEXDUMP_ROWPREFIXES.length) {
            dump.append(HEXDUMP_ROWPREFIXES[row]);
        } else {
            dump.append(StringUtil.NEWLINE);
            dump.append(Long.toHexString(rowStartIndex & 0xFFFFFFFFL | 0x100000000L));
            dump.setCharAt(dump.length() - 9, '|');
            dump.append('|');
        }
    }

    public static short getUnsignedByte(ByteBuffer buffer, int index) {
        return (short) (buffer.get(index) & 0xFF);
    }
}

测试类

java

package cn.itcast.netty.c1;

import java.nio.ByteBuffer;

import static cn.itcast.netty.c1.ByteBufferUtil.debugAll;

public class TestByteBufferReadWrite {

    public static void main(String[] args) {
        ByteBuffer buffer=ByteBuffer.allocate(10);
        buffer.put((byte) 0x61); // 'a'

        debugAll(buffer);

        buffer.put(new byte[]{0x62,0x63,0x64}); // b c d
        debugAll(buffer);

        buffer.flip();
        System.out.println(buffer.get());

        debugAll(buffer);

        buffer.compact();
        debugAll(buffer);
    }
}

结果如下

4) 字符串与ByteBuffer互转

方法一：

编码：字符串调用getByte方法获得byte数组，将byte数组放入ByteBuffer中

解码：先调用ByteBuffer的flip方法，然后通过StandardCharsets的decoder方法解码

java

package cn.itcast.netty.c1;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

public class Translate {
    public static void main(String[] args) {
        // 准备两个字符串
        String str1 = "hello";
        String str2 = "";


        ByteBuffer buffer1 = ByteBuffer.allocate(16);
        // 通过字符串的getByte方法获得字节数组，放入缓冲区中
        buffer1.put(str1.getBytes());
        ByteBufferUtil.debugAll(buffer1);

        // 将缓冲区中的数据转化为字符串
        // 切换模式
        buffer1.flip();
        
        // 通过StandardCharsets解码，获得CharBuffer，再通过toString获得字符串
        str2 = StandardCharsets.UTF_8.decode(buffer1).toString();
        System.out.println(str2);
        ByteBufferUtil.debugAll(buffer1);
    }
}

运行结果如下

java

+--------+-------------------- all ------------------------+----------------+
position: [5], limit: [16]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 65 6c 6c 6f 00 00 00 00 00 00 00 00 00 00 00 |hello...........|
+--------+-------------------------------------------------+----------------+
hello
+--------+-------------------- all ------------------------+----------------+
position: [5], limit: [5]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 65 6c 6c 6f 00 00 00 00 00 00 00 00 00 00 00 |hello...........|
+--------+-------------------------------------------------+----------------+

方法二

编码：通过StandardCharsets的encode方法获得ByteBuffer，此时获得的ByteBuffer为读模式，无需通过flip切换模式

解码：通过StandardCharsets的decoder方法解码

java

public class Translate {
    public static void main(String[] args) {
        // 准备两个字符串
        String str1 = "hello";
        String str2 = "";

        // 通过StandardCharsets的encode方法获得ByteBuffer
        // 此时获得的ByteBuffer为读模式，无需通过flip切换模式
        ByteBuffer buffer1 = StandardCharsets.UTF_8.encode(str1);
        ByteBufferUtil.debugAll(buffer1);

        // 将缓冲区中的数据转化为字符串
        // 通过StandardCharsets解码，获得CharBuffer，再通过toString获得字符串
        str2 = StandardCharsets.UTF_8.decode(buffer1).toString();
        System.out.println(str2);
        ByteBufferUtil.debugAll(buffer1);
    }
}

运行结果

txt

+--------+-------------------- all ------------------------+----------------+
position: [0], limit: [5]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 65 6c 6c 6f                                  |hello           |
+--------+-------------------------------------------------+----------------+
hello
+--------+-------------------- all ------------------------+----------------+
position: [5], limit: [5]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 65 6c 6c 6f                                  |hello           |
+--------+-------------------------------------------------+----------------+

方法三

编码：字符串调用getByte()方法获得字节数组，将字节数组传给ByteBuffer的wrap()方法，通过该方法获得ByteBuffer。同样无需调用flip方法切换为读模式

解码：通过StandardCharsets的decoder方法解码

public class Translate {
    public static void main(String[] args) {
        // 准备两个字符串
        String str1 = "hello";
        String str2 = "";

        // 通过StandardCharsets的encode方法获得ByteBuffer
        // 此时获得的ByteBuffer为读模式，无需通过flip切换模式
        ByteBuffer buffer1 = ByteBuffer.wrap(str1.getBytes());
        ByteBufferUtil.debugAll(buffer1);

        // 将缓冲区中的数据转化为字符串
        // 通过StandardCharsets解码，获得CharBuffer，再通过toString获得字符串
        str2 = StandardCharsets.UTF_8.decode(buffer1).toString();
        System.out.println(str2);
        ByteBufferUtil.debugAll(buffer1);
    }
}Copy

运行结果

+--------+-------------------- all ------------------------+----------------+
position: [0], limit: [5]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 65 6c 6c 6f                                  |hello           |
+--------+-------------------------------------------------+----------------+
hello
+--------+-------------------- all ------------------------+----------------+
position: [5], limit: [5]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 65 6c 6c 6f                                  |hello           |
+--------+-------------------------------------------------+----------------+

5) 分散读取

java

package cn.itcast.netty.c1;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

public class ScatteringReading {

    public static void main(String[] args) {
        try(RandomAccessFile file=new RandomAccessFile("3parts.txt","rw")) {
            FileChannel fileChannel=file.getChannel();
            ByteBuffer buffer1=ByteBuffer.allocate(3);
            ByteBuffer buffer2=ByteBuffer.allocate(3);
            ByteBuffer buffer3=ByteBuffer.allocate(5);

            fileChannel.read(new ByteBuffer[]{buffer1,buffer2,buffer3});

            buffer1.flip();
            buffer2.flip();
            buffer3.flip();

            ByteBufferUtil.debugAll(buffer1); //之前写的工具类
            ByteBufferUtil.debugAll(buffer2);//之前写的工具类
            ByteBufferUtil.debugAll(buffer3);//之前写的工具类
        }catch (IOException e){
            e.printStackTrace();
        }
    }
}

6) 粘包与半包

现象

网络上有多条数据发送给服务端，数据之间使用 \n 进行分隔但由于某种原因这些数据在接收时，被进行了重新组合，例如原始数据有3条为

Hello,world\n
I’m Nyima\n
How are you?\n

变成了下面的两个 byteBuffer (粘包，半包)

Hello,world\nI’m Nyima\nHo
w are you?\n

出现原因

粘包

发送方在发送数据时，并不是一条一条地发送数据，而是将数据整合在一起，当数据达到一定的数量后再一起发送。这就会导致多条信息被放在一个缓冲区中被一起发送出去

半包

接收方的缓冲区的大小是有限的，当接收方的缓冲区满了以后，就需要将信息截断，等缓冲区空了以后再继续放入数据。这就会发生一段完整的数据最后被截断的现象

究其根本原因，TCP为流式协议，消息不存在边界。

解决办法

通过get(index)方法遍历ByteBuffer，遇到分隔符时进行处理。
注意
：get(index)不会改变position的值
- 记录该段数据长度，以便于申请对应大小的缓冲区
- 将缓冲区的数据通过get()方法写入到target中
调用compact方法切换模式，因为缓冲区中可能还有未读的数据

java

package cn.itcast.netty.c1;

import java.nio.ByteBuffer;

public class ByteBufferDemo {
    public static void main(String[] args) {
        ByteBuffer buffer = ByteBuffer.allocate(32);
        // 模拟粘包+半包
        buffer.put("Hello,world\nI'm Nyima ".getBytes());
        // 调用split函数处理
        split(buffer);
        buffer.put("i am a girl?\nHo".getBytes());
        split(buffer);
        buffer.put("w are you?\n".getBytes());

        split(buffer);
    }

    private static void split(ByteBuffer buffer) {
        // 切换为读模式
        buffer.flip();
        for(int i = 0; i < buffer.limit(); i++) {

            // 遍历寻找分隔符
            // get(i)不会移动position
            if (buffer.get(i) == '\n') {
                // 缓冲区长度
                int length = i+1-buffer.position();
                ByteBuffer target = ByteBuffer.allocate(length);
                // 将前面的内容写入target缓冲区
                for(int j = 0; j < length; j++) {
                    // 将buffer中的数据写入target中
                    target.put(buffer.get());
                }
                // 打印查看结果
                ByteBufferUtil.debugAll(target);
            }
        }
        // 切换为写模式，但是缓冲区可能未读完，这里需要使用compact
        buffer.compact();
    }
}

运行结果

+--------+-------------------- all ------------------------+----------------+
position: [12], limit: [12]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 2c 77 6f 72 6c 64 0a             |Hello,world.    |
+--------+-------------------------------------------------+----------------+
+--------+-------------------- all ------------------------+----------------+
position: [23], limit: [23]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 49 27 6d 20 4e 79 69 6d 61 20 69 20 61 6d 20 61 |I'm Nyima i am a|
|00000010| 20 67 69 72 6c 3f 0a                            | girl?.         |
+--------+-------------------------------------------------+----------------+
+--------+-------------------- all ------------------------+----------------+
position: [13], limit: [13]
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 6f 77 20 61 72 65 20 79 6f 75 3f 0a          |How are you?.   |
+--------+-------------------------------------------------+----------------+

Process finished with exit code 0

解决方案：

1.固定长度法：服务端和客户端规定固定长度的缓冲区，当消息数据长度不足时，使用规定的填充字符进行填充。弊端：增加了不必要的数据传输，造成网络传输负担，不建议使用。

2.结束标识法：在包体尾部增加标识符表示一条完整的消息数据已经结束。弊端：若消息体本身包含该标识符需要做转义处理，因此效率依然不高。

3.长度信息法：将包体分为消息头+消息体，消息头中信息为消息体的长度，接收方通过该长度信息读取后面指定长度的内容，需要注意的是需限制可能的最大长度从而规定长度占用字节数。该方法为处理粘包半包问题的常用方法。核心代码：

3. 文件编程（了解即可）

3.1 FileChannel

获取

不能直接打开FileChannel，必须通过FilelnputStream、FileOutputStream或者RandomAccessFile 来获取FileChannel，它们都有getChannel方法

通过FilelnputStream 获取的channel 只能读。
通过 FileOutputStream 获取的channel 只能写
通过RandomAccessFile是否能读写根据构造RandomAccessFile时的读写模式决定

读取

会从channel读取数据填充ByteBuffer，返回值表示读到了多少字节，-1表示到达了文件的末尾

java

int readBytes = channel.read(buffer);

可根据返回值判断是否读取完毕

java

while(channel.read(buffer) > 0) {
    // 进行对应操作
    ...
}

写入

因为channel也是有大小的，所以 write 方法并不能保证一次将 buffer 中的内容全部写入 channel。必须需要按照以下规则进行写入

java

// 通过hasRemaining()方法查看缓冲区中是否还有数据未写入到通道中
while(buffer.hasRemaining()) {
	channel.write(buffer);
}

关闭

通道需要close，一般情况通过try-with-resource进行关闭，最好使用以下方法获取strea以及channel，避免某些原因使得资源未被关闭

java

public class TestChannel {
    public static void main(String[] args) throws IOException {
        try (FileInputStream fis = new FileInputStream("stu.txt");
             FileOutputStream fos = new FileOutputStream("student.txt");
             FileChannel inputChannel = fis.getChannel();
             FileChannel outputChannel = fos.getChannel()) {
            
            // 执行对应操作
            ...
                
        }
    }
}

位置

position

channel也拥有一个保存读取数据位置的属性，即position

java

long pos = channel.position();

可以通过position(int pos)设置channel中position的值

java

long newPos = ...;
channel.position(newPos);

设置当前位置时，如果设置为文件的末尾

这时读取会返回 -1
这时写入，会追加内容，但要注意如果 position 超过了文件末尾，再写入时在新内容和原末尾之间会有空洞（00）

强制写入

操作系统出于性能的考虑，会将数据缓存，不是立刻写入磁盘，而是等到缓存满了以后将所有数据一次性的写入磁盘。可以调用 force(true) 方法将文件内容和元数据（文件的权限等信息）立刻写入磁盘

3.2 两个Channel传输数据

transferTo方法

使用transferTo方法可以快速、高效地将一个channel中的数据传输到另一个channel中，但一次只能传输2G的内容

transferTo底层使用了零拷贝技术

java

package cn.itcast.netty.c1;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.channels.FileChannel;

public class TestChannel {
    public static void main(String[] args){
        try (FileInputStream fis = new FileInputStream("stu.txt");
             FileOutputStream fos = new FileOutputStream("student.txt");
             FileChannel inputChannel = fis.getChannel();
             FileChannel outputChannel = fos.getChannel()) {
            // 参数：inputChannel的起始位置，传输数据的大小，目的channel
            // 返回值为传输的数据的字节数
            // transferTo一次只能传输2G的数据
            inputChannel.transferTo(0, inputChannel.size(), outputChannel);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

当传输的文件大于2G时，需要使用以下方法进行多次传输

java

public class TestChannel {
    public static void main(String[] args){
        try (FileInputStream fis = new FileInputStream("stu.txt");
             FileOutputStream fos = new FileOutputStream("student.txt");
             FileChannel from = fis.getChannel();
             FileChannel to = fos.getChannel()) {
            long size = from.size();
            long capacity = from.size();
            // 分多次传输
            while (capacity > 0) {
                // transferTo返回值为传输了的字节数
                capacity -= from.transferTo(size-capacity, capacity, to);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

3.3 Path与Paths

Path 用来表示文件路径
Paths 是工具类，用来获取 Path 实例

java

Path source = Paths.get("1.txt"); // 相对路径 不带盘符 使用 user.dir 环境变量来定位 1.txt

Path source = Paths.get("d:\\1.txt"); // 绝对路径 代表了  d:\1.txt 反斜杠需要转义

Path source = Paths.get("d:/1.txt"); // 绝对路径 同样代表了  d:\1.txt

Path projects = Paths.get("d:\\data", "projects"); // 代表了  d:\data\projectsCopy

. 代表了当前路径
.. 代表了上一级路径

例如目录结构如下

d:
	|- data
		|- projects
			|- a
			|- b

代码

java

Path path = Paths.get("d:\\data\\projects\\a\\..\\b");
System.out.println(path);
System.out.println(path.normalize()); // 正常化路径 会去除 . 以及 ..

输出结果为

d:\data\projects\a\..\b
d:\data\projects\b

3.4 Files

查找

检查文件是否存在

java

Path path = Paths.get("helloword/data.txt");
System.out.println(Files.exists(path));

创建
创建一级目录
java
```
Path path = Paths.get("helloword/d1");
Files.createDirectory(path);
```
- 如果目录已存在，会抛异常 FileAlreadyExistsException
- 不能一次创建多级目录，否则会抛异常 NoSuchFileException
创建多级目录用
java
```
Path path = Paths.get("helloword/d1/d2");
Files.createDirectories(path);
```

拷贝文件

java

Path source = Paths.get("helloword/data.txt");
Path target = Paths.get("helloword/target.txt");

Files.copy(source, target);

如果文件已存在，会抛异常 FileAlreadyExistsException

如果希望用 source 覆盖掉 target，需要用 StandardCopyOption 来控制

java

Files.copy(source, target, StandardCopyOption.REPLACE_EXISTING);

移动文件

java

Path source = Paths.get("helloword/data.txt");
Path target = Paths.get("helloword/data.txt");

Files.move(source, target, StandardCopyOption.ATOMIC_MOVE);

StandardCopyOption.ATOMIC_MOVE 保证文件移动的原子性

删除
删除文件
java
```
Path target = Paths.get("helloword/target.txt");

Files.delete(target);
```
如果文件不存在，会抛异常 NoSuchFileException
删除目录
java
```
Path target = Paths.get("helloword/d1");

Files.delete(target);
```
如果目录还有内容，会抛异常 DirectoryNotEmptyException

遍历

可以使用Files工具类中的walkFileTree(Path, FileVisitor)方法，其中需要传入两个参数

Path：文件起始路径
FileVisitor：文件访问器，
使用访问者模式
- 接口的实现类
  SimpleFileVisitor
  有四个方法
  - preVisitDirectory：访问目录前的操作
  - visitFile：访问文件的操作
  - visitFileFailed：访问文件失败时的操作
  - postVisitDirectory：访问目录后的操作

Demo

遍历多级目录

java

package cn.itcast.netty.c1;

import java.io.IOException;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.concurrent.atomic.AtomicInteger;

public class TestWalkFileTree {
    public static void main(String[] args) throws IOException {
        Path path = Paths.get("D:\\maven");
        // 文件目录数目
        AtomicInteger dirCount = new AtomicInteger();
        // 文件数目
        AtomicInteger fileCount = new AtomicInteger();
        Files.walkFileTree(path, new SimpleFileVisitor<Path>(){

            // 在遍历文件夹之前
            @Override
            public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
                System.out.println("===>"+dir);
                // 增加文件目录数
                dirCount.incrementAndGet();
                return super.preVisitDirectory(dir, attrs);
            }
            // 拜访文件数量
            @Override
            public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
                System.out.println(file);
                // 增加文件数
                fileCount.incrementAndGet();
                return super.visitFile(file, attrs);
            }
        });
        // 打印数目
        System.out.println("文件目录数:"+dirCount.get());
        System.out.println("文件数:"+fileCount.get());
    }
}

运行结果如下

txt

...
===>F:\JDK 8\lib\security\policy\unlimited
F:\JDK 8\lib\security\policy\unlimited\local_policy.jar
F:\JDK 8\lib\security\policy\unlimited\US_export_policy.jar
F:\JDK 8\lib\security\trusted.libraries
F:\JDK 8\lib\sound.properties
F:\JDK 8\lib\tzdb.dat
F:\JDK 8\lib\tzmappings
F:\JDK 8\LICENSE
F:\JDK 8\README.txt
F:\JDK 8\release
F:\JDK 8\THIRDPARTYLICENSEREADME-JAVAFX.txt
F:\JDK 8\THIRDPARTYLICENSEREADME.txt
F:\JDK 8\Welcome.html
文件目录数:23
文件数:279

删除多级目录

java

    import java.io.IOException;
    import java.nio.file.*;
    import java.nio.file.attribute.BasicFileAttributes;
    
    /**
     * @author lilinchao
     * @date 2022/5/30
     * @description 删除多级目录
     **/
    public class FilesDeleteDirectoryDemo {
        public static void main(String[] args) throws IOException {
            Path path = Paths.get("D:\\libs2");
    
            Files.walkFileTree(path,new SimpleFileVisitor<Path>(){
                //visitFile()在访问任何文件时被调用
                @Override
                public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
                    Files.delete(file);
                    return super.visitFile(file, attrs);
                }
    
                //postVisitDirectory()在访问任何目录后被调用
                @Override
                public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
                    Files.delete(dir);
                    return super.postVisitDirectory(dir, exc);
                }
            });
            System.out.println("删除完毕!");
        }
    }

拷贝多级目录

java

    import java.io.IOException;
    import java.nio.file.Files;
    import java.nio.file.Paths;
    
    /**
     * @author lilinchao
     * @date 2022/5/30
     * @description 拷贝多级目录
     **/
    public class FilesCopyDirectoryDemo {
        public static void main(String[] args) throws IOException {
            long start = System.currentTimeMillis();
            String source = "D:\\libs";
            String target = "D:\\libs_aaa";
    
            Files.walk(Paths.get(source)).forEach(path -> {
                try {
                    String targetName = path.toString().replace(source,target);
                    //是目录
                    if(Files.isDirectory(path)){
                        //createDirectory()方法利用Path创建一个新的目录
                        Files.createDirectory(Paths.get(targetName));
                    }else if(Files.isRegularFile(path)){ //是普通文件
                        //copy()方法将文件从一个path复制到另一个
                        Files.copy(path,Paths.get(targetName));
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            long end = System.currentTimeMillis();
            System.out.println(end - start);
        }
    }

4. 网络编程

4.1 阻塞-BIO

阻塞模式下，相关方法都会导致线程暂停
- ServerSocketChannel.accept 会在没有连接建立时让线程暂停
- SocketChannel.read 会在通道中没有数据可读时让线程暂停
- 阻塞的表现其实就是线程暂停了，暂停期间不会占用 cpu，但线程相当于闲置
单线程下，阻塞方法之间相互影响，几乎不能正常工作，需要多线程支持
但多线程下，有新的问题，体现在以下方面
- 32 位 jvm 一个线程 320k，64 位 jvm 一个线程 1024k，如果连接数过多，必然导致 OOM，并且线程太多，反而会因为频繁上下文切换导致性能降低
- 可以采用线程池技术来减少线程数和线程上下文切换，但治标不治本，如果有很多连接建立，但长时间 inactive，会阻塞线程池中所有线程，因此不适合长连接，只适合短连接

服务端代码

java

package cn.itcast.netty.c1;

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;

public class Server {
    public static void main(String[] args) {
        // 创建缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 获得服务器通道
        try(ServerSocketChannel server = ServerSocketChannel.open()) {
            // 为服务器通道绑定端口
            server.bind(new InetSocketAddress(8080));
            // 用户存放连接的集合
            ArrayList<SocketChannel> channels = new ArrayList<>();
            // 循环接收连接
            while (true) {
                System.out.println("before connecting...");
                // 没有连接时，会阻塞线程
                SocketChannel socketChannel = server.accept();
                System.out.println("after connecting...");
                channels.add(socketChannel);
                // 循环遍历集合中的连接
                for(SocketChannel channel : channels) {
                    System.out.println("before reading");
                    // 处理通道中的数据
                    // 当通道中没有数据可读时，会阻塞线程
                    channel.read(buffer);
                    buffer.flip();
                    ByteBufferUtil.debugRead(buffer);
                    buffer.clear();
                    System.out.println("after reading");
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

客户端代码

java

package cn.itcast.netty.c1;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class Client {
    public static void main(String[] args) {
        try (SocketChannel socketChannel = SocketChannel.open()) {
            // 建立连接
            socketChannel.connect(new InetSocketAddress("localhost", 8080));
            System.out.println("waiting...");
            socketChannel.write(Charset.defaultCharset().encode("hello"));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

4.2 非阻塞-NIO

可以通过ServerSocketChannel的configureBlocking(false)方法将获得连接设置为非阻塞的。此时若没有连接，accept会返回null
可以通过SocketChannel的configureBlocking(false)方法将从通道中读取数据设置为非阻塞的。若此时通道中没有数据可读，read会返回-1

java

package cn.itcast.netty.c1;

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;

public class Server_Plus {
    public static void main(String[] args) {
        // 创建缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 获得服务器通道
        try (ServerSocketChannel server = ServerSocketChannel.open()) {
            // 为服务器通道绑定端口
            server.bind(new InetSocketAddress(8080));
            // 用户存放连接的集合
            ArrayList<SocketChannel> channels = new ArrayList<>();
            // 循环接收连接
            while (true) {
                // 设置为非阻塞模式，没有连接时返回null，不会阻塞线程
                server.configureBlocking(false);
                SocketChannel socketChannel = server.accept();
                // 通道不为空时才将连接放入到集合中
                if (socketChannel != null) {
                    System.out.println("after connecting...");
                    channels.add(socketChannel);
                }
                // 循环遍历集合中的连接
                for (SocketChannel channel : channels) {
                    // 处理通道中的数据
                    // 设置为非阻塞模式，若通道中没有数据，会返回0，不会阻塞线程
                    channel.configureBlocking(false);
                    int read = channel.read(buffer);
                    if (read > 0) {
                        buffer.flip();
                        ByteBufferUtil.debugRead(buffer);
                        buffer.clear();
                        System.out.println("after reading");
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

这样写存在一个问题，因为设置为了非阻塞，会一直执行while(true)中的代码，CPU一直处于忙碌状态，会使得性能变低，所以实际情况中不使用这种方法处理请求

4.3 Selector多路复用

单线程可以配合 Selector 完成对多个 Channel 可读写事件的监控，这称之为多路复用

多路复用仅针对网络 IO，普通文件 IO 无法利用多路复用
如果不用 Selector 的非阻塞模式，线程大部分时间都在做无用功，而 Selector 能够保证
- 有可连接事件时才去连接
- 有可读事件才去读取
- 有可写事件才去写入
  限于网络传输能力，Channel 未必时时可写，一旦 Channel 可写，会触发 Selector 的可写事件

以下是selector的四个事件。
accept事件
connect事件
read事件
write事件

4.4 accept事件

要使用Selector实现多路复用，服务端代码如下改进

java

public class SelectServer {
    public static void main(String[] args) {
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 获得服务器通道
        try(ServerSocketChannel server = ServerSocketChannel.open()) {
            server.bind(new InetSocketAddress(8080));
            // 创建选择器
            Selector selector = Selector.open();
            
            // 通道必须设置为非阻塞模式
            server.configureBlocking(false);
            // 将通道注册到选择器中，并设置感兴趣的事件
            server.register(selector, SelectionKey.OP_ACCEPT);
            while (true) {
                // 若没有事件就绪，线程会被阻塞，反之不会被阻塞。从而避免了CPU空转
                // 返回值为就绪的事件个数
                int ready = selector.select();
                System.out.println("selector ready counts : " + ready);
                
                // 获取所有事件
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                
                // 使用迭代器遍历事件
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    
                    // 判断key的类型
                    if(key.isAcceptable()) {
                        // 获得key对应的channel
                        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                        System.out.println("before accepting...");
                        
        				// 获取连接并处理，而且是必须处理，否则需要取消
                        SocketChannel socketChannel = channel.accept();
                        System.out.println("after accepting...");
                        
                        // 处理完毕后移除
                        iterator.remove();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

步骤解析

获得选择器Selector

java

Selector selector = Selector.open();

将通道设置为非阻塞模式，并注册到选择器中，并设置感兴趣的事件

channel 必须工作在非阻塞模式
FileChannel 没有非阻塞模式，因此不能配合 selector 一起使用
绑定的
事件类型
可以有
- connect - 客户端连接成功时触发
- accept - 服务器端成功接受连接时触发
- read - 数据可读入时触发，有因为接收能力弱，数据暂不能读入的情况
- write - 数据可写出时触发，有因为发送能力弱，数据暂不能写出的情况

java

// 通道必须设置为非阻塞模式
server.configureBlocking(false);
// 将通道注册到选择器中，并设置感兴趣的实践
server.register(selector, SelectionKey.OP_ACCEPT);

通过Selector监听事件，并获得就绪的通道个数，若没有通道就绪，线程会被阻塞

阻塞直到绑定事件发生
java
```
int count = selector.select();
```
阻塞直到绑定事件发生，或是超时（时间单位为 ms）
java
```
int count = selector.select(long timeout);
```
不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件
java
```
int count = selector.selectNow();
```

获取就绪事件并得到对应的通道，然后进行处理

java

// 获取所有事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();
                
// 使用迭代器遍历事件
Iterator<SelectionKey> iterator = selectionKeys.iterator();

while (iterator.hasNext()) {
	SelectionKey key = iterator.next();
                    
	// 判断key的类型，此处为Accept类型
	if(key.isAcceptable()) {
        // 获得key对应的channel
        ServerSocketChannel channel = (ServerSocketChannel) key.channel();

        // 获取连接并处理，而且是必须处理，否则需要取消
        SocketChannel socketChannel = channel.accept();

        // 处理完毕后移除
        iterator.remove();
	}
}

事件发生后能否不处理

事件发生后，要么处理，要么取消（cancel），不能什么都不做，否则下次该事件仍会触发，这是因为 nio 底层使用的是水平触发

4.5 Read事件

java

package cn.itcast.netty.c1;

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;

public class Server_Selector {

    public static void main(String[] args) throws IOException {

        // 1. 创建 selector，管理多个channel
        Selector selector = Selector.open();

        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false); // 设置非阻塞通道


        // 2. 建立 selector和channel的联系
        SelectionKey selectionKey = ssc.register(selector, 0, null);
        // key只关注accept事件
        selectionKey.interestOps(SelectionKey.OP_ACCEPT);
        System.out.println("register key: " + selectionKey);

        ssc.bind(new InetSocketAddress(8080)); // 绑定端口号
        ByteBuffer buffer = ByteBuffer.allocate(16);

        while (true) {

            // 3. select 方法，没有事件发生，线程阻塞，有事件，线程才会恢复运行
            selector.select();
            // 4. 处理事件,selectedKeys内部包含的所有发生的事件。
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();// accept,read

            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                System.out.println("key: " + key);

                //  key.cancel();
                // 5. 区分事件
                if (key.isAcceptable()) {
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();

                    System.out.println("before accepting...");
                    SocketChannel sc = serverSocketChannel.accept();
                    sc.configureBlocking(false);
                    System.out.println("after accepting...");

                    // 设置为非阻塞模式，同时将连接的通道也注册到选择其中
                    SelectionKey scKey = sc.register(selector, 0, null);
                    // 只关注 读取事件
                    scKey.interestOps(SelectionKey.OP_READ);
                    System.out.println("scKey: " + scKey);


                } else if (key.isReadable()) {

                    try {
                        // 如果是可读事件。
                        SocketChannel channel = (SocketChannel) key.channel();
                        // 设置为非阻塞模式，若通道中没有数据，会返回0，不会阻塞线程
                        channel.configureBlocking(false);

                        int read = channel.read(buffer);
                        if (read > 0) {
                            buffer.flip();
                            System.out.println("before reading...");
                            ByteBufferUtil.debugRead(buffer);
                            System.out.println("after reading...");
                            buffer.clear();
                        } else {
                            // 因为 key会一直产生read事件。所以需要手动的进行cancel.
                            key.cancel();
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                        key.cancel();
                    }
                }
                // 处理完毕后移除
                iterator.remove();
            }
        }
    }
}

删除事件

当处理完一个事件后，一定要调用迭代器的remove方法移除对应事件，否则会出现错误。原因如下

以我们上面的 Read事件 的代码为例

当调用了 server.register(selector, SelectionKey.OP_ACCEPT)后，Selector中维护了一个集合，用于存放SelectionKey以及其对应的通道

java

// WindowsSelectorImpl 中的 SelectionKeyImpl数组
private SelectionKeyImpl[] channelArray = new SelectionKeyImpl[8];

java

public class SelectionKeyImpl extends AbstractSelectionKey {
    // Key对应的通道
    final SelChImpl channel;
    ...
}

当选择器中的通道对应的事件发生后，selecionKey会被放到另一个集合中，但是selecionKey不会自动移除，所以需要我们在处理完一个事件后，通过迭代器手动移除其中的selecionKey。否则会导致已被处理过的事件再次被处理，就会引发错误

断开处理

当客户端与服务器之间的连接断开时，会给服务器端发送一个读事件，对异常断开和正常断开需要加以不同的方式进行处理

正常断开

正常断开时，服务器端的channel.read(buffer)方法的返回值为-1，所以当结束到返回值为-1时，需要调用key的cancel方法取消此事件，并在取消后移除该事件

java

int read = channel.read(buffer);
// 断开连接时，客户端会向服务器发送一个写事件，此时read的返回值为-1
if(read == -1) {
    // 取消该事件的处理
	key.cancel();
    channel.close();
} else {
    ...
}
// 取消或者处理，都需要移除key
iterator.remove();

异常断开
- 异常断开时，会抛出IOException异常，在try-catch的catch块中捕获异常并调用key的cancel方法即可

消息边界

不处理消息边界存在的问题--原因

将缓冲区的大小设置为4个字节，发送2个汉字（你好），

java

package cn.itcast.netty.c1;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class Client {
    public static void main(String[] args) {
        try (SocketChannel socketChannel = SocketChannel.open()) {
            // 建立连接
            socketChannel.connect(new InetSocketAddress("localhost", 8080));
            System.out.println("waiting...");
            socketChannel.write(Charset.defaultCharset().encode("中国"));

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

通过decode解码并打印时，会出现乱码

txt

ByteBuffer buffer = ByteBuffer.allocate(4);
// 解码并打印
System.out.println(StandardCharsets.UTF_8.decode(buffer));

结果

txt

中�
��

这是因为UTF-8字符集下，1个汉字占用3个字节，此时缓冲区大小为4个字节，一次读时间无法处理完通道中的所有数据，所以一共会触发两次读事件。这就导致 中国 的 中 字被拆分为了前半部分和后半部分发送，解码时就会出现问题

处理消息边界--解决办法

传输的文本可能有以下三种情况

文本大于缓冲区大小
- 此时需要将缓冲区进行扩容
发生半包现象
发生粘包现象

解决思路大致有以下三种

固定消息长度，数据包大小一样，服务器按预定长度读取，当发送的数据较少时，需要将数据进行填充，直到长度与消息规定长度一致。缺点是浪费带宽
另一种思路是按分隔符拆分，缺点是效率低，需要一个一个字符地去匹配分隔符
TLV 格式，即 Type 类型、Length 长度、Value 数据（也就是在消息开头用一些空间存放后面数据的长度），如HTTP请求头中的Content-Type与Content-Length类型和长度已知的情况下，就可以方便获取消息大小，分配合适的 buffer，缺点是 buffer 需要提前分配，如果内容过大，则影响 server 吞吐量。
- Http 1.1 是 TLV 格式
- Http 2.0 是 LTV 格式

下文的消息边界处理方式为第二种：按分隔符拆分

mermaid

sequenceDiagram
participant c1 as 客户端1
participant s as 服务器
participant b1 as ByteBuffer1
participant b2 as ByteBuffer2
c1 ->> s:发送0123456789abcdef3333\n
s ->>b1:第一次read存入0123456789abcdef
s ->>b2:扩容
b1 ->> b2:拷贝0123456789abcdef
s ->> b2:第二次read存入3333\n
b2 ->> b2: 0123456789abcdef3333\n

附件与扩容

Channel的register方法还有第三个参数：附件，可以向其中放入一个Object类型的对象，该对象会与登记的Channel以及其对应的SelectionKey绑定，可以从SelectionKey获取到对应通道的附件

java

public final SelectionKey register(Selector sel, int ops, Object att)

可通过SelectionKey的attachment()方法获得附件

java

ByteBuffer buffer = (ByteBuffer) key.attachment();

我们需要在Accept事件发生后，将通道注册到Selector中时，对每个通道添加一个ByteBuffer附件，让每个通道发生读事件时都使用自己的通道，避免与其他通道发生冲突而导致问题

java

// 设置为非阻塞模式，同时将连接的通道也注册到选择其中，同时设置附件
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(16);
// 添加通道对应的Buffer附件
socketChannel.register(selector, SelectionKey.OP_READ, buffer);

当Channel中的数据大于缓冲区时，需要对缓冲区进行扩容操作。此代码中的扩容的判定方法：Channel调用compact方法后，的position与limit相等，说明缓冲区中的数据并未被读取（容量太小），此时创建新的缓冲区，其大小扩大为两倍。同时还要将旧缓冲区中的数据拷贝到新的缓冲区中，同时调用SelectionKey的attach方法将新的缓冲区作为新的附件放入SelectionKey中

java

// 如果缓冲区太小，就进行扩容
if (buffer.position() == buffer.limit()) {
    ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity()*2);
    // 将旧buffer中的内容放入新的buffer中
    ewBuffer.put(buffer);
    // 将新buffer作为附件放到key中
    key.attach(newBuffer);
}

改造后的java代码

java

public class SelectServer {
    public static void main(String[] args) {
        // 获得服务器通道
        try(ServerSocketChannel server = ServerSocketChannel.open()) {
            server.bind(new InetSocketAddress(8080));
            // 创建选择器
            Selector selector = Selector.open();
            // 通道必须设置为非阻塞模式
            server.configureBlocking(false);
            // 将通道注册到选择器中，并设置感兴趣的事件
            server.register(selector, SelectionKey.OP_ACCEPT);
            // 为serverKey设置感兴趣的事件
            while (true) {
                // 若没有事件就绪，线程会被阻塞，反之不会被阻塞。从而避免了CPU空转
                // 返回值为就绪的事件个数
                int ready = selector.select();
                System.out.println("selector ready counts : " + ready);
                // 获取所有事件
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                // 使用迭代器遍历事件
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    // 判断key的类型
                    if(key.isAcceptable()) {
                        // 获得key对应的channel
                        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                        System.out.println("before accepting...");
                        // 获取连接
                        SocketChannel socketChannel = channel.accept();
                        System.out.println("after accepting...");
                        // 设置为非阻塞模式，同时将连接的通道也注册到选择其中，同时设置附件
                        socketChannel.configureBlocking(false);
                        ByteBuffer buffer = ByteBuffer.allocate(16);
                        socketChannel.register(selector, SelectionKey.OP_READ, buffer);
                        // 处理完毕后移除
                        iterator.remove();
                    } else if (key.isReadable()) {
                        SocketChannel channel = (SocketChannel) key.channel();
                        System.out.println("before reading...");
                        // 通过key获得附件（buffer）
                        ByteBuffer buffer = (ByteBuffer) key.attachment();
                        int read = channel.read(buffer);
                        if(read == -1) {
                            key.cancel();
                            channel.close();
                        } else {
                            // 通过分隔符来分隔buffer中的数据
                            split(buffer);
                            // 如果缓冲区太小，就进行扩容
                            if (buffer.position() == buffer.limit()) {
                                ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity()*2);
                                // 将旧buffer中的内容放入新的buffer中
                                buffer.flip();
                                newBuffer.put(buffer);
                                // 将新buffer放到key中作为附件
                                key.attach(newBuffer);
                            }
                        }
                        System.out.println("after reading...");
                        // 处理完毕后移除
                        iterator.remove();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void split(ByteBuffer buffer) {
        buffer.flip();
        for(int i = 0; i < buffer.limit(); i++) {
            // 遍历寻找分隔符
            // get(i)不会移动position
            if (buffer.get(i) == '\n') {
                // 缓冲区长度
                int length = i+1-buffer.position();
                ByteBuffer target = ByteBuffer.allocate(length);
                // 将前面的内容写入target缓冲区
                for(int j = 0; j < length; j++) {
                    // 将buffer中的数据写入target中
                    target.put(buffer.get());
                }
                // 打印结果
                ByteBufferUtil.debugAll(target);
            }
        }
        // 切换为写模式，但是缓冲区可能未读完，这里需要使用compact
        buffer.compact();
    }
}

发送端代码

java

package cn.itcast.netty.c1;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class Client {
    public static void main(String[] args) {
        try (SocketChannel socketChannel = SocketChannel.open()) {
            // 建立连接
            socketChannel.connect(new InetSocketAddress("localhost", 8080));
            System.out.println("waiting...");
            socketChannel.write(Charset.defaultCharset().encode("0123456789abcdef3333\nsd334\n"));

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

ByteBuffer的大小分配

每个 channel 都需要记录可能被切分的消息，因为 ByteBuffer 不能被多个 channel 共同使用，因此需要为每个 channel 维护一个独立的 ByteBuffer
ByteBuffer 不能太大，比如一个 ByteBuffer 1Mb 的话，要支持百万连接就要 1Tb 内存，因此需要设计大小可变的 ByteBuffer
分配思路可以参考
- 一种思路是首先分配一个较小的 buffer，例如 4k，如果发现数据不够，再分配 8k 的 buffer，将 4k buffer 内容拷贝至 8k buffer，优点是消息连续容易处理，缺点是数据拷贝耗费性能
  - 参考实现 http://tutorials.jenkov.com/java-performance/resizable-array.html
- 另一种思路是用多个数组组成 buffer，一个数组不够，把多出来的内容写入新的数组，与前面的区别是消息存储不连续解析复杂，优点是避免了拷贝引起的性能损耗

4.6 Write事件

服务器通过Buffer向通道中写入数据时，可能因为通道容量小于Buffer中的数据大小，导致无法一次性将Buffer中的数据全部写入到Channel中，这时便需要分多次写入，具体步骤如下

执行一次写操作，向将buffer中的内容写入到SocketChannel中，然后判断Buffer中是否还有数据

若Buffer中还有数据，则需要将SockerChannel注册到Seletor中，并关注写事件，同时将未写完的Buffer作为附件一起放入到SelectionKey中

java

 int write = socket.write(buffer);
// 通道中可能无法放入缓冲区中的所有数据
if (buffer.hasRemaining()) {
    // 注册到Selector中，关注可写事件，并将buffer添加到key的附件中
    socket.configureBlocking(false);
    socket.register(selector, SelectionKey.OP_WRITE, buffer);
}

添加写事件的相关操作key.isWritable()，对Buffer再次进行写操作

每次写后需要判断Buffer中是否还有数据（是否写完）。若写完，需要移除SelecionKey中的Buffer附件，避免其占用过多内存，同时还需移除对写事件的关注

java

SocketChannel socket = (SocketChannel) key.channel();
// 获得buffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
// 执行写操作
int write = socket.write(buffer);
System.out.println(write);
// 如果已经完成了写操作，需要移除key中的附件，同时不再对写事件感兴趣
if (!buffer.hasRemaining()) {
    key.attach(null);
    key.interestOps(0);
}

整体代码如下

java

public class WriteServer {
    public static void main(String[] args) {
        try(ServerSocketChannel server = ServerSocketChannel.open()) {
            server.bind(new InetSocketAddress(8080));
            server.configureBlocking(false);
            Selector selector = Selector.open();
            server.register(selector, SelectionKey.OP_ACCEPT);
            while (true) {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    // 处理后就移除事件
                    iterator.remove();
                    if (key.isAcceptable()) {
                        // 获得客户端的通道
                        SocketChannel socket = server.accept();
                        // 写入数据
                        StringBuilder builder = new StringBuilder();
                        for(int i = 0; i < 500000000; i++) {
                            builder.append("a");
                        }
                        ByteBuffer buffer = StandardCharsets.UTF_8.encode(builder.toString());
                        // 先执行一次Buffer->Channel的写入，如果未写完，就添加一个可写事件
                        int write = socket.write(buffer);
                        System.out.println(write);
                        // 通道中可能无法放入缓冲区中的所有数据
                        if (buffer.hasRemaining()) {
                            // 注册到Selector中，关注可写事件，并将buffer添加到key的附件中
                            socket.configureBlocking(false);
                            socket.register(selector, SelectionKey.OP_WRITE, buffer);
                        }
                    } else if (key.isWritable()) {
                        SocketChannel socket = (SocketChannel) key.channel();
                        // 获得buffer
                        ByteBuffer buffer = (ByteBuffer) key.attachment();
                        // 执行写操作
                        int write = socket.write(buffer);
                        System.out.println(write);
                        // 如果已经完成了写操作，需要移除key中的附件，同时不再对写事件感兴趣
                        if (!buffer.hasRemaining()) {
                            key.attach(null);
                            key.interestOps(0);
                        }
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

接受数据的端口

java

package cn.itcast.netty.c1;

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

public class WriteClient {
    public static void main(String[] args) throws IOException {
        SocketChannel sc=SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost",8080));

        int count=0;
        // 3. 接受
        while (true){
            ByteBuffer buffer=ByteBuffer.allocate(1024*1024);
            count+=sc.read(buffer);
            System.out.println(count);
            buffer.clear();
        }


    }
}

4.7 多线程优化

充分利用多核CPU，分两组选择器

单线程配一个选择器（Boss），专门处理 accept 事件
创建 cpu 核心数的线程（Worker），每个线程配一个选择器，轮流处理 read 事件

实现思路

创建一个负责处理Accept事件的Boss线程，与多个负责处理Read事件的Worker线程

Boss线程执行的操作

接受并处理Accepet事件，当Accept事件发生后，调用Worker的register(SocketChannel socket)方法，让Worker去处理Read事件，其中需要根据标识robin去判断将任务分配给哪个Worker

java

// 创建固定数量的Worker
Worker[] workers = new Worker[4];
// 用于负载均衡的原子整数
AtomicInteger robin = new AtomicInteger(0);
// 负载均衡，轮询分配Worker
workers[robin.getAndIncrement()% workers.length].register(socket);

register(SocketChannel socket)方法会通过同步队列完成Boss线程与Worker线程之间的通信，让SocketChannel的注册任务被Worker线程执行。添加任务后需要调用selector.wakeup()来唤醒被阻塞的Selector

java

public void register(final SocketChannel socket) throws IOException {
    // 只启动一次
    if (!started) {
       // 初始化操作
    }
    // 向同步队列中添加SocketChannel的注册事件
    // 在Worker线程中执行注册事件
    queue.add(new Runnable() {
        @Override
        public void run() {
            try {
                socket.register(selector, SelectionKey.OP_READ);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    });
    // 唤醒被阻塞的Selector
    // select类似LockSupport中的park，wakeup的原理类似LockSupport中的unpark
    selector.wakeup();
}

Worker线程执行的操作
- 从同步队列中获取注册任务，并处理Read事件

java代码

java

package cn.itcast.netty.c1;

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;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadsServer {
    public static void main(String[] args) {
        try (ServerSocketChannel server = ServerSocketChannel.open()) {
            // 当前线程为Boss线程
            Thread.currentThread().setName("Boss");
            server.bind(new InetSocketAddress(8080));
            // 负责轮询Accept事件的Selector
            Selector boss = Selector.open();
            server.configureBlocking(false);
            server.register(boss, SelectionKey.OP_ACCEPT);
            // 创建固定数量的Worker
            Worker[] workers = new Worker[4];
            // 用于负载均衡的原子整数
            AtomicInteger robin = new AtomicInteger(0);
            for (int i = 0; i < workers.length; i++) {
                workers[i] = new Worker("worker-" + i);
            }
            while (true) {
                boss.select();
                Set<SelectionKey> selectionKeys = boss.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    // BossSelector负责Accept事件
                    if (key.isAcceptable()) {
                        // 建立连接
                        SocketChannel socket = server.accept();
                        System.out.println("connected...");
                        socket.configureBlocking(false);
                        // socket注册到Worker的Selector中
                        System.out.println("before read...");
                        // 负载均衡，轮询分配Worker
                        workers[robin.getAndIncrement() % workers.length].register(socket);
                        System.out.println("after read...");
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    static class Worker implements Runnable {
        private Thread thread;
        private volatile Selector selector;
        private final String name;
        private volatile boolean started = false;
        /**
         * 同步队列，用于Boss线程与Worker线程之间的通信
         */
        private ConcurrentLinkedQueue<Runnable> queue;

        public Worker(String name) {
            this.name = name;
        }

        public void register(final SocketChannel socket) throws IOException {
            // 只启动一次
            if (!started) {
                thread = new Thread(this, name);
                selector = Selector.open();
                queue = new ConcurrentLinkedQueue<>();
                thread.start();
                started = true;
            }

            // 向同步队列中添加SocketChannel的注册事件
            // 在Worker线程中执行注册事件
            queue.add(new Runnable() {
                @Override
                public void run() {
                    try {
                        socket.register(selector, SelectionKey.OP_READ);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });
            // 唤醒被阻塞的Selector
            // select类似LockSupport中的park，wakeup的原理类似LockSupport中的unpark
            selector.wakeup();
        }

        @Override
        public void run() {
            while (true) {
                try {
                    selector.select();
                    // 通过同步队列获得任务并运行
                    Runnable task = queue.poll();
                    if (task != null) {
                        // 获得任务，执行注册操作
                        task.run();
                    }
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        // Worker只负责Read事件
                        if (key.isReadable()) {
                            // 简化处理，省略细节
                            SocketChannel socket = (SocketChannel) key.channel();
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            socket.read(buffer);
                            buffer.flip();
                            ByteBufferUtil.debugAll(buffer);
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

5. NIO vs BIO

5.1 Stream与Channel

stream 不会自动缓冲数据，channel 会利用系统提供的发送缓冲区、接收缓冲区（更为底层）
stream 仅支持阻塞 API，channel 同时支持阻塞、非阻塞 API，网络 channel 可配合 selector 实现多路复用
二者均为全双工，即读写可以同时进行
虽然Stream是单向流动的，但是它也是全双工的

5.2 IO模型

同步：线程自己去获取结果（一个线程）
例如：线程调用一个方法后，需要等待方法返回结果
异步：线程自己不去获取结果，而是由其它线程返回结果（至少两个线程）
例如：线程A调用一个方法后，继续向下运行，运行结果由线程B返回

当调用一次 channel.read 或 stream.read 后，会由用户态切换至操作系统内核态来完成真正数据读取，而读取又分为两个阶段，分别为：

等待数据阶段
复制数据阶段

根据UNIX 网络编程 - 卷 I，IO模型主要有以下几种

阻塞IO

用户线程进行read操作时，需要等待操作系统执行实际的read操作，此期间用户线程是被阻塞的，无法执行其他操作

非阻塞IO

用户线程在一个循环中一直调用read方法，若内核空间中还没有数据可读，立即返回
- 只是在等待阶段非阻塞
用户线程发现内核空间中有数据后，等待内核空间执行复制数据，待复制结束后返回结果

多路复用

ava中通过Selector实现多路复用

当没有事件是，调用select方法会被阻塞住
一旦有一个或多个事件发生后，就会处理对应的事件，从而实现多路复用

多路复用与阻塞IO的区别

阻塞IO模式下，若线程因accept事件被阻塞，发生read事件后，仍需等待accept事件执行完成后，才能去处理read事件
多路复用模式下，一个事件发生后，若另一个事件处于阻塞状态，不会影响该事件的执行

异步IO

线程1调用方法后理解返回，不会被阻塞也不需要立即获取结果
当方法的运行结果出来以后，由线程2将结果返回给线程1

信号驱动

5.3 零拷贝

零拷贝指的是数据无需拷贝到 JVM 内存中，同时具有以下三个优点

更少的用户态与内核态的切换
不利用 cpu 计算，减少 cpu 缓存伪共享
零拷贝适合小文件传输

传统 IO 问题

传统的 IO 将一个文件通过 socket 写出

java

File f = new File("helloword/data.txt");
RandomAccessFile file = new RandomAccessFile(file, "r");

byte[] buf = new byte[(int)f.length()];
file.read(buf);

Socket socket = ...;
socket.getOutputStream().write(buf);

内部工作流如下

Java 本身并不具备 IO 读写能力，因此 read 方法调用后，要从 Java 程序的用户态切换至内核态，去调用操作系统（Kernel）的读能力，将数据读入内核缓冲区。这期间用户线程阻塞，操作系统使用 DMA（Direct Memory Access）来实现文件读，其间也不会使用 CPU
DMA 也可以理解为硬件单元，用来解放 cpu 完成文件 IO
从内核态切换回用户态，将数据从内核缓冲区读入用户缓冲区（即 byte[] buf），这期间 CPU 会参与拷贝，无法利用 DMA
调用 write 方法，这时将数据从用户缓冲区（byte[] buf）写入 socket 缓冲区，CPU 会参与拷贝
接下来要向网卡写数据，这项能力 Java 又不具备，因此又得从用户态切换至内核态，调用操作系统的写能力，使用 DMA 将 socket 缓冲区的数据写入网卡，不会使用 CPU

可以看到中间环节较多，java 的 IO 实际不是物理设备级别的读写，而是缓存的复制，底层的真正读写是操作系统来完成的

用户态与内核态的切换发生了 3 次，这个操作比较重量级
数据拷贝了共 4 次

NIO 优化

方法得通过 DirectByteBuf
ByteBuffer.allocate(10)
底层对应 HeapByteBuffer，使用的还是 Java 内存
ByteBuffer.allocateDirect(10)
底层对应DirectByteBuffer，使用的是操作系统内存

大部分步骤与优化前相同，唯有一点：Java 可以使用 DirectByteBuffer 将堆外内存映射到 JVM 内存中来直接访问使用

这块内存不受 JVM 垃圾回收的影响，因此内存地址固定，有助于 IO 读写
Java 中的 DirectByteBuf 对象仅维护了此内存的虚引用，内存回收分成两步
- DirectByteBuffer 对象被垃圾回收，将虚引用加入引用队列
  - 当引用的对象ByteBuffer被垃圾回收以后，虚引用对象Cleaner就会被放入引用队列中，然后调用Cleaner的clean方法来释放直接内存
  - DirectByteBuffer 的释放底层调用的是 Unsafe 的 freeMemory 方法
- 通过专门线程访问引用队列，根据虚引用释放堆外内存
减少了一次数据拷贝，用户态与内核态的切换次数没有减少

进一步优化1

以下两种方式都是零拷贝，即无需将数据拷贝到用户缓冲区中（JVM内存中）
底层采用了 linux 2.1 后提供的 sendFile 方法，Java 中对应着两个 channel 调用 transferTo/transferFrom 方法拷贝数据

Java 调用 transferTo 方法后，要从 Java 程序的用户态切换至内核态，使用 DMA将数据读入内核缓冲区，不会使用 CPU
数据从内核缓冲区传输到 socket 缓冲区，CPU 会参与拷贝
最后使用 DMA 将 socket 缓冲区的数据写入网卡，不会使用 CPU

这种方法下

只发生了1次用户态与内核态的切换
数据拷贝了 3 次

进一步优化2

linux 2.4 对上述方法再次进行了优化

Java 调用 transferTo 方法后，要从 Java 程序的用户态切换至内核态，使用 DMA将数据读入内核缓冲区，不会使用 CPU
只会将一些 offset 和 length 信息拷入 socket 缓冲区，几乎无消耗
使用 DMA 将 内核缓冲区的数据写入网卡，不会使用 CPU

整个过程仅只发生了1次用户态与内核态的切换，数据拷贝了 2 次

5.4 AIO

AIO 用来解决数据复制阶段的阻塞问题

同步意味着，在进行读写操作时，线程需要等待结果，还是相当于闲置
异步意味着，在进行读写操作时，线程不必等待结果，而是将来由操作系统来通过回调方式由另外的线程来获得结果

异步模型需要底层操作系统（Kernel）提供支持
Windows 系统通过 IOCP 实现了真正的异步 IO
Linux 系统异步 IO 在 2.6 版本引入，但其底层实现还是用多路复用模拟了异步 IO，性能没有优势

java

package cn.itcast.netty.c1;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;

public class AioFileChannel {

    public static void main(String[] args) throws InterruptedException {
        try (AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get("data.txt"), StandardOpenOption.READ)) {

            ByteBuffer buffer=ByteBuffer.allocate(16);
            // 参数1 文件放入的位置
            // 参数2 读取的起始位置
            // 参数3 附件
            // 参数4 回调对象 CompletionHandler
            channel.read(buffer, 0, buffer, new CompletionHandler<Integer, ByteBuffer>() {
                @Override
                public void completed(Integer result, ByteBuffer attachment) {
                    // 执行过程中成功
                    attachment.flip();
                    ByteBufferUtil.debugAll(attachment);
                }

                @Override
                public void failed(Throwable exc, ByteBuffer attachment) {
                    // 执行方法失败
                }
            });

            System.out.println("read end...");
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 这里需要睡眠的原因，因为 主线程结束，守护线程也会提前结束。
        Thread.sleep(1000L);
    }
}

Netty全套课程 ​

基础 ​

1. 三大组件 ​

Channel ​

Buffer ​

Selector ​

2. ByteBuffer ​

1) 入门Demo ​

2) ByteBuffer内部结构 ​

属性介绍 ​

结构 ​

常见方法 ​

put() ​

flip() ​

get() ​

rewind() ​

clean() ​

mark()/reset() ​

compact() ​

3) 方法调用及演示 ​

调试工具类(重要) ​

测试类 ​

4) 字符串与ByteBuffer互转 ​

5) 分散读取 ​

6) 粘包与半包 ​

现象 ​

出现原因 ​

解决办法 ​

3. 文件编程（了解即可） ​

3.1 FileChannel ​

3.2 两个Channel传输数据 ​

3.3 Path与Paths ​

3.4 Files ​

Demo ​

4. 网络编程 ​

4.1 阻塞-BIO ​

4.2 非阻塞-NIO ​

4.3 Selector多路复用 ​

4.4 accept事件 ​

4.5 Read事件 ​

删除事件 ​

断开处理 ​

消息边界 ​

ByteBuffer的大小分配 ​

4.6 Write事件 ​

4.7 多线程优化 ​

实现思路 ​

5. NIO vs BIO ​

5.1 Stream与Channel ​

5.2 IO模型 ​

阻塞IO ​

非阻塞IO ​

多路复用 ​

异步IO ​

信号驱动 ​

5.3 零拷贝 ​

传统 IO 问题 ​

NIO 优化 ​

进一步优化1 ​

进一步优化2 ​

5.4 AIO ​

Netty全套课程

基础

1. 三大组件

Channel

Buffer

Selector

2. ByteBuffer

1) 入门Demo

2) ByteBuffer内部结构

属性介绍

结构

常见方法

put()

flip()

get()

rewind()

clean()

mark()/reset()

compact()

3) 方法调用及演示

调试工具类(重要)

测试类

4) 字符串与ByteBuffer互转

5) 分散读取

6) 粘包与半包

现象

出现原因

解决办法

3. 文件编程（了解即可）

3.1 FileChannel

3.2 两个Channel传输数据

3.3 Path与Paths

3.4 Files

Demo

4. 网络编程

4.1 阻塞-BIO

4.2 非阻塞-NIO

4.3 Selector多路复用

4.4 accept事件

4.5 Read事件

删除事件

断开处理

消息边界

ByteBuffer的大小分配

4.6 Write事件

4.7 多线程优化

实现思路

5. NIO vs BIO

5.1 Stream与Channel

5.2 IO模型

阻塞IO

非阻塞IO

多路复用

异步IO

信号驱动

5.3 零拷贝

传统 IO 问题

NIO 优化

进一步优化1

进一步优化2

5.4 AIO