NIO 基础

三大组件

Channel & Buffer & Selector

channel 读写数据是双向通道，类似于 stream，不过 inputstram是只能读，outputstream 只能输出。

常见的 Channel：

• FileChannel：文件传输通道
• DatagramChannel：UDP 传输通道
• SocketChannel：TCP传输通道，客户端服务端都可以用
• ServerSocketChannel：专用服务器 TCP传输通道

buffer 则用来缓冲读写数据，常见的 buffer 有：

• ByteBuffer（最为常用）
  • MappedByteBuffer
  • DirectByteBuffer
  • HeapByteBuffer
• ShortBuffer
• IntBuffer
• LongBuffer
• FloatBuffer
• DoubleBuffer
• CharBuffer

在nio之前啊，开发一个服务器端的程序，采用多线程的方式

每连接一个客户端，就会创建一个线程去对接。

• 缺点：内存占用高
• 线程上下文切换成本太高
• 线程在等待的时候，需要的时候再恢复，这就是线程上下文切换
• 只能适合连接数少的情况下

线程池可以实现线程复用，但是存在阻塞情况

线程池版本也可以设计，会阻塞。一个线程对接多个客户端，只能一个一个来，这就阻塞了。

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

我们看看 Selector 是怎么设计的？

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

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

bytebuffer 的基本使用

所需要依赖

<!--netty 网络编程-->
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.39.Final</version>
</dependency>
<!--JSON 转换-->
<dependency>
    <groupId>com.google.code.gson</groupId>
    <artifactId>gson</artifactId>
    <version>2.8.5</version>
</dependency>
<!--goole 工具类-->
<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>19.0</version>
</dependency>
<!--日志输出调试-->
<dependency>
    <groupId>ch.qos.logback</groupId>
    <artifactId>logback-classic</artifactId>
    <version>1.2.3</version>
</dependency>

做一个文件读取

ByteBuffer 正确使用姿势

1. 初始状态是空的，也就是只能写入
2. 向 buffer 写入数据，例如调用 channel.read(buffer)
3. 调用 flip() 切换至读模式（不切换读不到数据）
4. 从 buffer 读取数据，例如调用 buffer.get()
5. 调用 clear() 或 compact() 切换至写模式
6. 重复 2~5 步骤

// 读取一个文件获取通道
try (FileChannel channel = new FileInputStream("src/data.txt").getChannel()) {
    ByteBuffer buffer = ByteBuffer.allocate(10);
    while (true) {
        // 从 channel 读取数据，向 buffer 写入
        // 准备缓冲区，指定大小，10个字节
        int len = channel.read(buffer);
        // 读取到的字节数 = len
        if (len == -1) { // 没有内容了
            break;
        }
        // 打印 buffer 的内容
        buffer.flip(); // 切换读模式
        while (buffer.hasRemaining()) { // 是否还要剩余未读的
            byte b = buffer.get();
            System.out.println((char) b);
        }
        buffer.clear(); // 切换写模式
    }

} catch (IOException ignored) {
}

bytebuffer 结构

• capacity：容量，代表能装多少数据
• position：读写指针，索引下标，读到哪里写到哪里
• limit：读写限制，应该读多少字节，写多少字节

看看执行过程

1. 一开始

1. 写模式下，position 是写入位置，limit 等于容量，下图表示写入了 4 个字节后的状态

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

1. 读取 4 个字节后，状态

1. clear 动作发生后，状态

1. compact 方法，是把未读完的部分向前压缩，然后切换至写模式（用在未读完的情况）

ByteBuffer 代码分析

工具类

package com.example;

import io.netty.util.internal.StringUtil;

import java.nio.ByteBuffer;

import static io.netty.util.internal.MathUtil.isOutOfBounds;
import static io.netty.util.internal.StringUtil.NEWLINE;

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(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 (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(
                "         +-------------------------------------------------+" +
                        NEWLINE + "         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |" +
                        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(NEWLINE +
                "+--------+-------------------------------------------------+----------------+");
    }

    private static void appendHexDumpRowPrefix(StringBuilder dump, int row, int rowStartIndex) {
        if (row < HEXDUMP_ROWPREFIXES.length) {
            dump.append(HEXDUMP_ROWPREFIXES[row]);
        } else {
            dump.append(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);
    }
}

代码测试

public static void main(String[] args) {
    // 分配 10字节 空间
    ByteBuffer buffer = ByteBuffer.allocate(10);
    // 向 buffer 写入数据
    buffer.put((byte) 0x61);
    // 查看信息
    debugAll(buffer);
    // 写入数据
    buffer.put(new byte[]{0x62, 0x63, 0x64});
    debugAll(buffer);
    // 没有转读模式，是无法读取的
//        System.out.println(buffer.get());
    buffer.flip();// 切换读模式
    System.out.println((char) buffer.get());
    debugAll(buffer);
    buffer.compact();
    // 为什么我们将未读入的元素往前移动了，还是有数据在原位置呢
    // 虽然原位置移动了，但是下一个写入的位置一定是这个元素之前的
    // 所以这个位置一定会被重新写入
    debugAll(buffer);
    buffer.put(new byte[]{0x65, 0x66});
    debugAll(buffer);
}

常见方法（分配空间，写数据）

可以使用 allocate 方法为 ByteBuffer 分配空间

public static void main(String[] args) {
    // 垃圾回收：当java虚拟机内存满是，触发垃圾回收
    // 通过垃圾回收算法内存碎片整理化，减少内存碎片，让内存更紧凑、
    // 内存移动进行搬迁重新赋值

    // Java 堆内存，读写效率较低，受到 GC（垃圾回收影响）
    ByteBuffer.allocate(16);
    // 直接内存：读写效率较高（少一次拷贝），不会受到 GC 影响，分配的效率慢（需要掉系统函数，进行分配内存） // 使用不当内存泄露
    ByteBuffer.allocateDirect(16);
}

向 buffer 写入数据

有两种办法

• 调用 channel 的 read 方法
• 调用 buffer 自己的 put 方法

方式一

// 创建 buffer，分配空间
ByteBuffer buffer = ByteBuffer.allocateDirect(16);
// 获取通道
FileChannel channel = new FileInputStream("data.txt").getChannel();
// 读入数据
channel.read(buffer);

方式二

ByteBuffer buffer = ByteBuffer.allocateDirect(16);
buffer.put((byte) 0x22);

从 buffer 读取数据

同样有两种办法

• 调用 channel 的 write 方法
• 调用 buffer 自己的 get 方法

方式一

public static void main(String[] args) {
    // 分配空间
    ByteBuffer buffer = ByteBuffer.allocateDirect(16);
    // 获取通道
    FileChannel channel = new FileInputStream("data.txt").getChannel();
    // 写入数据
    channel.write(buffer);
}

方式二

public static void main(String[] args) {
    // 分配空间
    ByteBuffer buffer = ByteBuffer.allocateDirect(16);
    buffer.get(); // 读入一个数据，将指针指向下一个
//        buffer.rewind(); // 将指针重置为0
//        buffer.get(i); // i 为索引，不会移动指针
}

演示一下，标记位置，将指针重置到标记点

public static void main(String[] args) {
    // 分配空间
    ByteBuffer buffer = ByteBuffer.allocateDirect(10);
    // 写入数据
    buffer.put(new byte[]{'a', 'b', 'c', 'd', 'e'});
    buffer.flip();// 切换读模式

    buffer.get(new byte[4]); // 读四个
    debugAll(buffer);

    // 从头开始读
    buffer.rewind();
    System.out.println((char) buffer.get()); // a
    // mark(做标记) & reset(将指针重置到标记的位置)
    System.out.println((char) buffer.get()); // b
    System.out.println((char) buffer.get()); // c
    buffer.mark(); // 标记
    System.out.println((char) buffer.get()); // d
    System.out.println((char) buffer.get()); // e
    buffer.reset(); // 指针重置到标记点
    System.out.println((char) buffer.get()); // d
    System.out.println((char) buffer.get()); // e
    // get(i) 不会改变索引位置
    System.out.println((char) buffer.get(3));
    debugAll(buffer);
}

字符串与 ByteBuffer 互转

public static void main(String[] args) {
    // 1. 字符串转为 getBytes，这种方式转完是写模式
    ByteBuffer buffer = ByteBuffer.allocateDirect(10);
    buffer.put("hello".getBytes());
    debugAll(buffer);

    // 2. Charset--自动切换读模式
    ByteBuffer buffer2 = StandardCharsets.UTF_8.encode("hello");
    debugAll(buffer2);

    // 3. wrap---自动切换读模式
    ByteBuffer buff3 = ByteBuffer.wrap("hello".getBytes());
    debugAll(buff3);

    // ByteBuffer 转字符串，只能在读模式下，写模式无法转
    String str1 = StandardCharsets.UTF_8.decode(buffer2).toString();
    System.out.println(str1);

    // 写模式下
    String str2 = StandardCharsets.UTF_8.decode(buffer).toString();
    System.out.println(str2);
}

Buffer 是非线程安全的

Scattering Reads & Gathering Writes(分散度集中写)

现在有一个文本文件（one two three），我要读取进来

第一种思路，全部读取到一个 byteBuffer里，然后再拆分成三组

1. 做一次完整的读取
2. 再进行拆分，涉及数据的重新分割复制

第二种思路，如果数据是已知的，读取时一次分别读到三个byteBuffer 里

• 这就叫做Scattering Reads

分散读

public static void main(String[] args) {
    try (FileChannel r = new RandomAccessFile("src/data.txt", "r").getChannel()) {
        ByteBuffer byteBuffer = ByteBuffer.allocateDirect(3);
        ByteBuffer byteBuffer2 = ByteBuffer.allocateDirect(3);
        ByteBuffer byteBuffer3 = ByteBuffer.allocateDirect(5);
        r.read(new ByteBuffer[]{byteBuffer, byteBuffer2, byteBuffer3});
        byteBuffer.flip();
        byteBuffer2.flip();
        byteBuffer3.flip();
        debugAll(byteBuffer); // 查看读取信息
        debugAll(byteBuffer2);
        debugAll(byteBuffer3);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

集中写

public static void main(String[] args) {
    ByteBuffer b1 = StandardCharsets.UTF_8.encode("hello");
    ByteBuffer b2 = StandardCharsets.UTF_8.encode("world");
    ByteBuffer b3 = StandardCharsets.UTF_8.encode("你好");
    try (FileChannel chancel = new RandomAccessFile("src/data.txt", "rw").getChannel()) {
        chancel.write(new ByteBuffer[]{b1, b2, b3});
    } catch (IOException e) {
    }
}

优点：可以减少数据在 byteBuffer 之间的拷贝，减少数据复制次数，变相提高效率

黏包 / 半包

黏包：多个数据发送过来，变成了一个行

半包：多个数据在被服务端接收到，后面的数据换行了

发送的数据

• Hello,world\n
• I'm zhangsan\n
• How are you?\n

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

• Hello,world\nI'm zhangsan\nHo
• w are you?\n

原因：为了提高效率，会存在缓冲区，将多个数据同时放入缓冲区发送给服务端，由于缓冲区大小存在限制，导致出现半包。

例子

public class TestByteBuffer {
    public static void main(String[] args) {
        ByteBuffer source = ByteBuffer.allocate(32);
        source.put("hello,world\nI'm zhangsan\nHo".getBytes());
        split(source);
        source.put("w are you?\n".getBytes());
        split(source);
    }

    private static void split(ByteBuffer source) {
        source.flip();
        // 遍历到最大的limit
        for (int i = 0; i < source.limit(); i++) {
            if (source.get(i) == '\n') {
                // 计算读取到的字节数
                int length = i + 1 - source.position();
                ByteBuffer newBuffer = ByteBuffer.allocate(length);
                for (int j = 0; j < length; j++) {
                    newBuffer.put(source.get());
                }
                debugAll(newBuffer);
            }
        }
        source.compact(); // 清除掉已读的
    }
}

文件编程

FileChannel 只能工作在阻塞模式下，它不能和Selector一起用

获取 & 读取 & 写入

获取

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

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

读取

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

int readBytes = channel.read(buffer);

写入

写入的正确姿势如下

ByteBuffer buffer = ...;
buffer.put(...); // 存入数据
buffer.flip();   // 切换读模式

while(buffer.hasRemaining()) { // 判断是否有下一个数据
    channel.write(buffer);
}

在 while 中调用 channel.write 是因为 write 方法并不能保证一次将 buffer 中的内容全部写入 channel

& 关闭 & 位置 & 大小 & 强制写入

关闭

channel 必须关闭，不过调用了 FileInputStream、FileOutputStream 或者 RandomAccessFile 的 close 方法会间接地调用 channel 的 close 方法

位置

// 获取当前位置
long pos = channel.position();
// 设置当前位置
long newPos = ...;
channel.position(newPos);

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

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

大小

使用 size 方法获取文件的大小

强制写入

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

两个 Channel 传输数据

transferTo：返回值是读取了多少数据字节数，最大只能读取 2G

效率高，底层hui利用操作系统的零拷贝进行优化

public static void main(String[] args) {
    try (
            FileChannel form = new FileInputStream("D:/这个夏天/CentOS 7.zip").getChannel();
            FileChannel to = new FileOutputStream("src/to.zip").getChannel()
    ) {
        // 利用 transferTo 进行复制数据
        long size = form.size(); // 全部字节
        long left = size; // 可读取的字节数
        while (left >= 1) {
            // 每次减去已经读取的数量
            // size - left ，起始位置等于刚刚读取的位置
            left -= form.transferTo(size - left, form.size(), to);
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}

常用方法 Path

jdk7 引入了 Path 和 Paths 类

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

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\projects

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

常用方法 Files

检查文件是否存在

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

创建一级目录

Path path = Paths.get("helloword/d1");
Files.createDirectory(path);

• 如果目录已存在，会抛异常 FileAlreadyExistsException
• 不能一次创建多级目录，否则会抛异常 NoSuchFileException

创建多级目录用

Path path = Paths.get("helloword/d1/d2");
Files.createDirectories(path);

拷贝文件

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

Files.copy(source, target);

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

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

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

移动文件

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

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

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

删除文件

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

Files.delete(target);

如果文件不存在，会抛异常 NoSuchFileException

删除目录

Path target = Paths.get("helloword/d1");

Files.delete(target);

如果目录还有内容，会抛异常 DirectoryNotEmptyException

目录文件操作（Files.walkFileTree & Files.walk）

• visitFileFailed：遍历时异常解决方法
• visitFile：文件方法
• preVisitDirectory：目录方法
• postVisitDirectory：目录退出的方法

记录总数

public static void main(String[] args) throws IOException {
    Path path = Paths.get("D:\WorkSoftWare\Navicat Premium 16");
    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); // 目录总数
    System.out.println(fileCount); // 文件总数
}

统计 jar 的数目

Path path = Paths.get("C:\\Program Files\\Java\\jdk1.8.0_91");
AtomicInteger fileCount = new AtomicInteger();
Files.walkFileTree(path, new SimpleFileVisitor<Path>(){
    @Override
    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) 
        throws IOException {
        if (file.toFile().getName().endsWith(".jar")) {
            fileCount.incrementAndGet();
        }
        return super.visitFile(file, attrs);
    }
});
System.out.println(fileCount); // 724

删除多级目录

1. 先进入目录，如果存在文件，则删除
2. 退出目录时删除目录

public static void main(String[] args) throws IOException {
    Path path = Paths.get("D:\\WorkSoftWare\\Navicat Premium 16 - 副本");
    AtomicInteger dirCount = new AtomicInteger();
    AtomicInteger fileCount = new AtomicInteger();
    Files.walkFileTree(path, new SimpleFileVisitor<Path>() {

        // 文件的方法
        @Override
        public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
            Files.delete(file); // 删除文件
            return super.visitFile(file,attrs);
        }

        @Override
        public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
            Files.delete(dir); // 退出时删除目录
            return super.postVisitDirectory(dir, exc);
        }
    });
}

拷贝多级目录

public static void main(String[] args) throws IOException {
    String source = "D:\WorkSoftWare\Navicat Premium 16";
    String target = "D:\WorkSoftWare\Navicat Premium 16--副本";
    Files.walk(Paths.get(source)).forEach(path -> {
        // 将当前文件的前面的路径替换为我们拷贝的新目录
        String targetName = path.toString().replace(source, target);
        try {
            if (Files.isDirectory(path)) {
                // 如果是目录就根据文件名称创建目录
                Files.createDirectory(Paths.get(targetName));
            } else if (Files.isRegularFile(path)) {
                // 是文件就直接复制了
                Files.copy(path, Paths.get(targetName));
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    });
}

网络编程

单线程阻塞模式

阻塞模式下，相关方法都会导致线程暂停

• ServerSocketChannel.accept 会在没有连接建立时让线程暂停
• SocketChannel.read 会在没有数据可读时让线程暂停
• 阻塞的表现其实就是线程暂停了，暂停期间不会占用 cpu，但线程相当于闲置

单线程下，阻塞方法之间相互影响，几乎不能正常工作，需要多线程支持

但多线程下，有新的问题，体现在以下方面

• 32 位 jvm 一个线程 320k，64 位 jvm 一个线程 1024k，如果连接数过多，必然导致 OOM，并且线程太多，反而会因为频繁上下文切换导致性能降低
• 可以采用线程池技术来减少线程数和线程上下文切换，但治标不治本，如果有很多连接建立，但长时间 inactive，会阻塞线程池中所有线程，因此不适合长连接，只适合短连接

服务端

public static void main(String[] args) throws IOException {
    // 使用 nio 来处理阻塞模式，单线程
    ByteBuffer buffer = ByteBuffer.allocate(16);
    // 创建服务器
    ServerSocketChannel ssc = ServerSocketChannel.open();
    // 绑定监听端口
    ssc.bind(new InetSocketAddress(8080));
    // 连接集合
    ArrayList<SocketChannel> channels = new ArrayList<>();
    while (true) {
        //  accept 建立与客户端连接，SocketChannel 用来与客户端之间通信
        log.debug("connecting...");
        SocketChannel sc = ssc.accept(); // 阻塞方法--会等待客户端链接
        log.debug("connected...{}", sc);
        channels.add(sc);
        for (SocketChannel channel : channels) {
            // 接受胡客户端发送的数据
            log.debug("before read...{}", channel);
            channel.read(buffer); // 阻塞等待客户端发送信息
            buffer.flip();
            debugRead(buffer);
            buffer.clear();
            log.debug("after read....{}", channel);
        }
    }
}

客户端

技巧：向服务端发送数据，调试模式下右键 Alt+f8 评估，输入一下内容

sc.write(Charset.defaultCharset().encode("hello"));

// 客户端
public class Client {
    public static void main(String[] args) throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 8080));
        System.out.println("waiting");
    }
}

那么再发送数据之后我还可以继续发送数据吗？ 不能当然不能了，会执行到等待链接语句，进入阻塞状态

总结：在执行一个方法的时候，会阻塞其他方法

非阻塞模式

非阻塞模式下，相关方法都会不会让线程暂停

• 在 ServerSocketChannel.accept 在没有连接建立时，会返回 null，继续运行
• SocketChannel.read 在没有数据可读时，会返回 0，但线程不必阻塞，可以去执行其它 SocketChannel 的 read 或是去执行 ServerSocketChannel.accept
• 写数据时，线程只是等待数据写入 Channel 即可，无需等 Channel 通过网络把数据发送出去

但非阻塞模式下，即使没有连接建立，和可读数据，线程仍然在不断运行，白白浪费了 cpu

数据复制过程中，线程实际还是阻塞的（AIO 改进的地方）

客户端不变

// 使用 nio 来理解非阻塞模式, 单线程
// 0. ByteBuffer
ByteBuffer buffer = ByteBuffer.allocate(16);
// 1. 创建了服务器
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false); // 非阻塞模式
// 2. 绑定监听端口
ssc.bind(new InetSocketAddress(8080));
// 3. 连接集合
List<SocketChannel> channels = new ArrayList<>();
while (true) {
    // 4. accept 建立与客户端连接， SocketChannel 用来与客户端之间通信
    SocketChannel sc = ssc.accept(); // 非阻塞，线程还会继续运行，如果没有连接建立，但sc是null
    if (sc != null) {
        log.debug("connected... {}", sc);
        sc.configureBlocking(false); // 非阻塞模式
        channels.add(sc);
    }
    for (SocketChannel channel : channels) {
        // 5. 接收客户端发送的数据
        int read = channel.read(buffer);// 非阻塞，线程仍然会继续运行，如果没有读到数据，read 返回 0
        if (read > 0) {
            buffer.flip();
            debugRead(buffer);
            buffer.clear();
            log.debug("after read...{}", channel);
        }
    }
}

监听 Channel 事件

可以通过下面三种方法来监听是否有事件发生，方法的返回值代表有多少 channel 发生了事件

方法1，阻塞直到绑定事件发生

int count = selector.select();

方法2，阻塞直到绑定事件发生，或是超时（时间单位为 ms）

int count = selector.select(long timeout);

方法3，不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件

int count = selector.selectNow();

select 何时不阻塞

事件发生时

• 客户端发起连接请求，会触发 accept 事件
• 客户端发送数据过来，客户端正常、异常关闭时，都会触发 read 事件，另外如果发送的数据大于 buffer 缓冲区，会触发多次读取事件
• channel 可写，会触发 write 事件
• 在 linux 下 nio bug 发生时
• 调用 selector.wakeup()
• 调用 selector.close()
• selector 所在线程 interrupt

多路复用 selector

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

保证：

• 有可连接事件时才去连接
• 有可读事件才去读取
• 有可写事件才去写入
• 限于网络传输能力，Channel 未必时时可写，一旦 Channel 可写，会触发 Selector 的可写事件

客户端还是之前的

说一下，监听事件的几个类型

• accept：会在连接时触发
• connect：连接建立后触发
• read：可读事件
• write：可写事件

执行步骤：

1. 创建 Selector管理
2. 把 channel 注册到 selector上
3. 调用 select 进入阻塞，有事件才执行
4. 处理事件

public static void main(String[] args) throws IOException {
    // 创建 selector 管理多个 channel
    Selector selector = Selector.open();
    ServerSocketChannel ssc = ServerSocketChannel.open();
    ssc.configureBlocking(false); // false，表示不阻塞，默认true

    // 建立 selector 和 channel 的联系（注册）
    // SelectionKey 发生事件时，通过它可以知道事件和那个 channel有关
    // ops：参数二，主要关注那个事件
    SelectionKey sscKey = ssc.register(selector, 0, null);
    log.debug("sscKey: {}", sscKey);
    // 会有连接需求时触发
    sscKey.interestOps(SelectionKey.OP_ACCEPT);
    ssc.bind(new InetSocketAddress(8082)); // 监听8082端口
    while (true) {
        // 没有事件时进入阻塞状态，有事件才会恢复运行
        selector.select();
        // 处理事件 SelectionKey中包含所有的发生的事件
        Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
        while (iter.hasNext()) {
            SelectionKey key = iter.next();
            log.debug("key: {}", key); // 拿到的是 sscKey，只有这一个
            // 拿到是那个channel 发生的事件
            ServerSocketChannel channel = (ServerSocketChannel) key.channel();
            // 进行建立连接
            SocketChannel sc = channel.accept();
            log.debug("{}", sc);
        }
    }
}

上面这情况是处理了事件，下次就会阻塞，如果我们不处理，就会一直不阻塞，进入循环

如果不想处理，就取消这个事件

selector.select(); 方法在事件未处理不会进行阻塞，不想处理可以取消这个事件

while (iter.hasNext()) {
    SelectionKey key = iter.next();
    log.debug("key: {}", key); // 拿到的是
    // 拿到是那个channel 发生的事件
    //  ServerSocketChannel channel = (ServerSocketChannel) key.channel();
    //  进行建立连接
    //  SocketChannel sc = channel.accept();
    //  log.debug("{}", sc);
    // 取消这个事件
    key.cancel();
}

我们来整理一下思路，前面代码是怎么运行的

1. 首先创建Selector
2. 创建连接Channel，来处理连接事件
3. 放入Selector中
4. 到while循环中，触发Select方法，进行阻塞等待事件
5. 当有事件来了，获取所有的事件keys
6. 这个集合中，只会加入事件，不会移除事件
7. 依次遍历，遍历之前调用迭代器直接删除当前key，再进行区分事件类型
8. 如果是连接事件， 进行连接
9. 再次注册一个读取事件，交给Selector，现在Selector管理两个事件了
10. 结束时，一定要移除事件
11. 如果是读入事件，执行读入。

重点移除事件

演示上边的思路

public static void main(String[] args) throws IOException {
    // 创建 selector 管理多个 channel
    Selector selector = Selector.open();
    ServerSocketChannel ssc = ServerSocketChannel.open();
    ssc.configureBlocking(false); // false，表示不阻塞，默认true

    // 建立 selector 和 channel 的联系（注册）
    // SelectionKey 发生事件时，通过它可以知道事件和那个 channel有关
    // ops：参数二，主要关注那个事件
    SelectionKey sscKey = ssc.register(selector, 0, null);
    log.debug("sscKey: {}", sscKey);
    // 会有连接需求时触发
    sscKey.interestOps(SelectionKey.OP_ACCEPT);
    ssc.bind(new InetSocketAddress(8082)); // 监听8080端口
    while (true) {
        // 没有事件时进入阻塞状态，有事件才会恢复运行
        selector.select();
        // 处理事件 SelectionKey中包含所有的发生的事件
        Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
        while (iter.hasNext()) {
            SelectionKey key = iter.next();
            log.debug("key: {}", key); // 拿到的是

            // 拿到key直接删掉
            iter.remove();

            // 区分事件类型
            if (key.isAcceptable()) { // 如果是accept
                ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                SocketChannel sc = channel.accept();
                log.debug("{}", sc);
                sc.configureBlocking(false);
                SelectionKey scKey = sc.register(selector, 0, null);
                scKey.interestOps(SelectionKey.OP_READ);
            } else if (key.isReadable()) { // 如果是read
                // 读入数据的
                SocketChannel channel = (SocketChannel) key.channel();
                ByteBuffer buffer = ByteBuffer.allocate(16);
                channel.read(buffer);
                buffer.flip();
                debugAll(buffer);
            }
        }
    }
}

现在当客户端强行关闭，或者正常关闭，出现异常问题

强行关闭，触发异常导致虚拟机停止：需要捕捉 IO 异常 正常关闭：触发读操作，读到了 -1

// 区分事件类型
if (key.isAcceptable()) { // 如果是accept
    // 未修改......
} else if (key.isReadable()) { // 如果是read
    try {
        // 读入数据的
        SocketChannel channel = (SocketChannel) key.channel();
        ByteBuffer buffer = ByteBuffer.allocate(16);
        int read = channel.read(buffer);
        // 如果是 -1 表示客户端要取消连接
        if (read == -1) {
            key.cancel();
        } else {
            buffer.flip();
            debugAll(buffer);
        }
    } catch (IOException e) {
        e.printStackTrace();
        // 客户端触发了强行关闭
        key.cancel();
    }
}

处理消息边界

当传入中文的时候，会导致乱码问题。因为读取的字节不是一个完整的字

一种思路是固定消息长度，数据包大小一样，服务器按预定长度读取，缺点是浪费带宽

另一种思路是按分隔符拆分，缺点是效率低

TLV 格式，即 Type 类型、Length 长度、Value 数据，类型和长度已知的情况下，就可以方便获取消息大小，分配合适的 buffer，缺点是 buffer 需要提前分配，如果内容过大，则影响 server 吞吐量

• Http 1.1 是 TLV 格式
• Http 2.0 是 LTV 格式

第三种就是，先读取一个字节，表示后面的数据需要多少空间，再读取一个字节，再分配指定空间

也可以先读取类型，再读取长度，再读取数据

演示一下，第二个，按分隔符拆分及其扩容的方式，第三个后面Netty中讲

逻辑：当 ByteBuffer 空间不足了，就进行扩容，然后自动再次触发读

1. 先把ByteBuffer 存入到key的附件中
2. 当空间满了，直接扩容，再存入到附件

服务端

// 区分事件类型
if (key.isAcceptable()) { // 如果是accept
    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
    SocketChannel sc = channel.accept();
    log.debug("{}", sc);
    sc.configureBlocking(false);
    // 将一个ByteBuffer 作为附件关联到 SelectionKey 上
    ByteBuffer buffer = ByteBuffer.allocate(16);
    SelectionKey scKey = sc.register(selector, 0, buffer);
    scKey.interestOps(SelectionKey.OP_READ);
} else if (key.isReadable()) { // 如果是read
    try {
        // 读入数据的
        SocketChannel channel = (SocketChannel) key.channel();
        // 获取 SelectionKey 上关联的附件
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        int read = channel.read(buffer);
        if (read == -1) {
            key.cancel();
        } else {
            split(buffer);
            if(buffer.position() == buffer.limit()){
                // 代表需要扩容，如果满满的，那么是没有办法进行压缩空间的，再spilt方法最后
                ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
                buffer.flip(); // 进行读入
                newBuffer.put(buffer); // 将数据转移
                key.attach(newBuffer); // 重新添加附件
            }
        }
    } catch (IOException e) {
        e.printStackTrace();
        key.cancel();
    }
}

客户端

public static void main(String[] args) throws IOException, IOException {
    SocketChannel sc = SocketChannel.open();
    sc.connect(new InetSocketAddress("localhost", 8082));
    sc.write(Charset.defaultCharset().encode("0123456879abcdef3333\n"));
    System.in.read(); // 阻塞方法
}

当前只能扩容，还应该再缩小的

ByteBuffer 大小分配

每个 channel 都需要记录可能被切分的消息，因为 ByteBuffer 不能被多个 channel 共同使用，因此需要为每个 channel 维护一个独立的 ByteBuffer

ByteBuffer 不能太大，比如一个 ByteBuffer 1Mb 的话，要支持百万连接就要 1Tb 内存，因此需要设计大小可变的 ByteBuffer

• 一种思路是首先分配一个较小的 buffer，例如 4k，如果发现数据不够，再分配 8k 的 buffer，将 4k buffer 内容拷贝至 8k buffer，优点是消息连续容易处理，缺点是数据拷贝耗费性能
• 另一种思路是用多个数组组成 buffer，一个数组不够，把多出来的内容写入新的数组，与前面的区别是消息存储不连续解析复杂，优点是避免了拷贝引起的性能损耗

以上全代码

public static void main(String[] args) throws IOException {
    // 创建 selector 管理多个 channel
    Selector selector = Selector.open();
    ServerSocketChannel ssc = ServerSocketChannel.open();
    ssc.configureBlocking(false); // false，表示不阻塞，默认true

    // 建立 selector 和 channel 的联系（注册）
    // SelectionKey 发生事件时，通过它可以知道事件和那个 channel有关
    // ops：参数二，主要关注那个事件
    SelectionKey sscKey = ssc.register(selector, 0, null);
    log.debug("sscKey: {}", sscKey);
    // 会有连接需求时触发
    sscKey.interestOps(SelectionKey.OP_ACCEPT);
    ssc.bind(new InetSocketAddress(8082)); // 监听8080端口
    while (true) {
        // 没有事件时进入阻塞状态，有事件才会恢复运行
        selector.select();
        // 处理事件 SelectionKey中包含所有的发生的事件
        Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
        while (iter.hasNext()) {
            SelectionKey key = iter.next();
            log.debug("key: {}", key); // 拿到的是

            // 拿到key直接删掉
            iter.remove();

            // 区分事件类型
            if (key.isAcceptable()) { // 如果是accept
                ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                SocketChannel sc = channel.accept();
                log.debug("{}", sc);
                sc.configureBlocking(false);
                // 将一个ByteBuffer 作为附件关联到 SelectionKey 上
                ByteBuffer buffer = ByteBuffer.allocate(16);
                SelectionKey scKey = sc.register(selector, 0, buffer);
                scKey.interestOps(SelectionKey.OP_READ);
            } else if (key.isReadable()) { // 如果是read
                try {
                    // 读入数据的
                    SocketChannel channel = (SocketChannel) key.channel();
                    // 获取 SelectionKey 上关联的附件
                    ByteBuffer buffer = (ByteBuffer) key.attachment();
                    int read = channel.read(buffer);
                    if (read == -1) {
                        key.cancel();
                    } else {
                        split(buffer);
                        if(buffer.position() == buffer.limit()){
                            // 代表需要扩容，如果满满的，那么是没有办法进行压缩空间的，再spilt方法最后
                            ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
                            buffer.flip(); // 进行读入
                            newBuffer.put(buffer); // 将数据转移
                            key.attach(newBuffer); // 重新添加附件
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                    key.cancel();
                }
            }
        }
    }
}

private static void split(ByteBuffer source) {
    source.flip();
    // 遍历到最大的limit
    for (int i = 0; i < source.limit(); i++) {
        if (source.get(i) == '\n') {
            // 计算读取到的字节数
            int length = i + 1 - source.position();
            ByteBuffer newBuffer = ByteBuffer.allocate(length);
            for (int j = 0; j < length; j++) {
                newBuffer.put(source.get());
            }
            debugAll(newBuffer);
        }
    }
    source.compact(); // 清楚掉已读的
}

可写事件

服务端

public static void main(String[] args) throws IOException {
    ServerSocketChannel ssc = ServerSocketChannel.open();
    ssc.configureBlocking(false);
    Selector selector = Selector.open();
    ssc.register(selector, SelectionKey.OP_ACCEPT);
    ssc.bind(new InetSocketAddress(8082));
    while (true) {
        selector.select();
        Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
        while (iter.hasNext()) {
            SelectionKey key = iter.next();
            iter.remove();
            if (key.isAcceptable()) {
                // 因为我们用 key.channel() 获取的也是 ssc 这个对象，直接就可以拿到它
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                // 这里并没有进行交给 Selector管理

                // 1. 向客户端发送大量数据
                StringBuilder sb = new StringBuilder();
                for (int i = 0; i < 300000000; i++) {
                    sb.append("a");
                }
                ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());
                // 判断是否还有可读字节数
                while (buffer.hasRemaining()){
                    // 返回值代表实际写入的字节数
                    int write = sc.write(buffer);
                    System.out.println(write);
                }
            }
        }
    }
}

客户端

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

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

总结：这种方式传输大量的数据，会造成堵塞，因为大量数据分开传递，会有缓冲区满的问题，造成 0 发送，下面看一个完善的方式

如果写缓冲区满了，可以去读操作

while (iter.hasNext()) {
    SelectionKey key = iter.next();
    iter.remove();
    if (key.isAcceptable()) {
        // 因为我们用 key.channel() 获取的也是 ssc 这个对象，直接就可以拿到它
        SocketChannel sc = ssc.accept();
        sc.configureBlocking(false);
        // 这里交给 Selector管理
        SelectionKey sckey = sc.register(selector, SelectionKey.OP_READ, null);

        // 1. 向客户端发送大量数据
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < 300000000; i++) {
            sb.append("a");
        }
        ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());

        // 返回值代表实际写入的字节数
        int write = sc.write(buffer);
        System.out.println(write);

        // 判断是否还有可读字节数
        if (buffer.hasRemaining()) {
            // 关注可写事件, sckey.interestOps() + 事件，这样不会覆盖
            sckey.interestOps(sckey.interestOps() + SelectionKey.OP_WRITE);
            sckey.attach(buffer);
        }
    }else if(key.isWritable()){ // 如果是写事件
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        SocketChannel sc = (SocketChannel) key.channel();
        int write = sc.write(buffer);
        System.out.println(write);
        // 清理操作
        if(!buffer.hasRemaining()){
            key.attach(null);
            // 不需要关注可写事件
            // key.interestOps() - SelectionKey.OP_WRITE 也行，取消可写事件
            key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
        }
    }
}

多线程优化

设计思想

现在都是多核 cpu，设计时要充分考虑别让 cpu 的力量被白白浪费

前面的代码只有一个选择器，没有充分利用多核 cpu，如何改进呢？

分两组选择器

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

说下执行步骤

• 首先建立链接
• 进行初始化selector，线程等
• 往队列加任务
• 加任务的时候，work0就启动了
• 刚开始没有事件就会进入堵塞
• 但是队列下面就进行了唤醒
• 进行获取队列，并执行注册
• 现在没有任务，所以进入循环再次堵塞
• 有数据发送过来，然后往下执行

服务端

@Slf4j
public class Server {

    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        Selector boss = Selector.open();
        SelectionKey bossKey = ssc.register(boss, 0, null);
        bossKey.interestOps(SelectionKey.OP_ACCEPT);
        ssc.bind(new InetSocketAddress(8082));

        // 创建多个work，并初始化
        Worker[] workers = new Worker[2];
        // 动态获取CPU核心数----jdk10才修复，在docker容器下拿到的是物理CPU个数
        int cpuCount = Runtime.getRuntime().availableProcessors();
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new Worker("worker-" + i);
        }

        // 链接建立了就创建work是不对的，创建固定的work
        // 一个work对应多个socketChanel
        // 建立多个链接都与work关联 Worker worker = new Worker("work-0");

        // 创建计数器
        AtomicInteger index = new AtomicInteger();
        while (true) {
            boss.select();
            Iterator<SelectionKey> iter = boss.selectedKeys().iterator();
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                iter.remove();
                if (key.isAcceptable()) {
                    // 因为我们用 key.channel() 获取的也是 ssc 这个对象，直接就可以拿到它
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);
                    // 客户端端口地址
                    log.debug("connected...{}", sc.getRemoteAddress());
                    // 2. 关联 Selector，不在关联boss，而是work中的selector
                    log.debug("before register...{}", sc.getRemoteAddress());

                    // 修改---1  sc代码运行在boss中，而select方法在work中，有刚才的问题。
                    // 下面的register在work中不就可以了嘛
                    // 移动过去，register执行完了就会执行 sc
                    workers[index.getAndIncrement() % workers.length].register(sc); // 多个轮训
//                    worker.register(sc); // 注册,这个要矮着下面的一个

                    log.debug("after register...{}", sc.getRemoteAddress());
                }
            }
        }
    }

    static class Worker implements Runnable {
        private Thread thread;
        private Selector selector;
        private String name;
        // 判断 register 是否执行过
        // 线程和selector是否已经初始化
        private volatile boolean start = false;
        // 想让代码在某个位置执行，我们使用消息对列
        private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();

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

        // 初始化线程和 Selector
        // 修改---2
        public void register(SocketChannel sc) throws IOException {
            // 每次调用创建一个线程，这个线程数，一个work只用一线程，不能每次new
            if (!start) {
                this.selector = Selector.open(); // Selector 也是只用一个
                // work执行到这里才会执行新线程
                thread = new Thread(this, name);
                thread.start();
                start = true;
            }
            // 向队列添加了任务，并没有立即执行
            queue.add(() -> {
                try {
                    sc.register(selector, SelectionKey.OP_READ, null);
                } catch (ClosedChannelException e) {
                    e.printStackTrace();
                }
            });
            selector.wakeup(); // 唤醒 select方法
        }

        @Override
        public void run() {
            while (true) {
                try {
                    // 一定向调用Selector再执行消息队列
                    this.selector.select();
                    // 获取队列
                    Runnable task = queue.poll();
                    // 如果队列不为空，执行
                    if (task != null) {
                        task.run(); // 在这个时候执行注册，肯定是在work中
                    }
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        if (key.isReadable()) {
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            SocketChannel channel = (SocketChannel) key.channel();
                            log.debug("reading...{}", channel.getRemoteAddress());
                            channel.read(buffer);
                            buffer.flip();
                            debugAll(buffer);
                        }
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}

客户端

public class Client {
    public static void main(String[] args) throws IOException {
        SocketChannel channel = SocketChannel.open();
        channel.connect(new InetSocketAddress("localhost", 8082));
        channel.write(Charset.defaultCharset().encode("123456"));
        System.in.read();
    }
}

Runtime.getRuntime().availableProcessors() 如果工作在 docker 容器下，因为容器不是物理隔离的，会拿到物理 cpu 个数，而不是容器申请时的个数

• 这个问题直到 jdk 10 才修复，使用 jvm 参数 UseContainerSupport 配置， 默认开启

IO 模型

stream vs channel

stream 不会自动缓冲数据，channel 会利用系统提供的发送缓冲区、接收缓冲区（更为底层）

stream 仅支持阻塞 API，channel 同时支持阻塞、非阻塞 API，网络 channel 可配合 selector 实现多路复用

二者均为全双工，即读写可以同时进行

unix 网络编程-卷1 听说很好，从事网络编程建议看看

阻塞IO

• 当调用一次chancel.read之后，会切换到操作系统来读取数据
• 等待数据阶段：等待客户端发送数据（阻塞）
• 复制数据阶段：将网卡复制到内存里返回，切换用户空间。执行结束

非阻塞IO

• 调用 chancle.read 之后，切换到操作系统进行读取数据
• 发现没有数据，立即返回，再次尝试读取数据
• 当某一次读取到了数据，进行数据复制（这里还是会阻塞）

复杂情况下多路复用与阻塞io

阻塞IO

• 发起一个 read事件，需要等待数据
• 复制数据，可能会调用accept
• 然后去建立连接，如果是Chancel2
• 如果chancel1 如果发生了read读数据，这里会产生阻塞
• 只能等待建立完连接，才可以read

多路复用

• select就能检测多个事件，select等待事件
• 检测到了，c1 的read事件，c2的连接事件，c1的read事件
• select会把这些事件都得到，一次返回多个事件，然后拿到SelectKeys
• 然后在循环中全部处理掉

好处，不用等，触发即有数据

同步、异步

同步阻塞、同步非阻塞、多路复用（同步）

异步阻塞（没有此情况）、异步非阻塞（异步）

• 同步：线程自己去获取结果（一个线程）
• 异步：线程自己不去获取结果，而是由其它线程送结果（至少两个线程）

传统 IO问题，零拷贝

将服务器的文件读取进来，通过socket api发送出去

先看看下面的代码

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 是没有能力去读取文件的，需要调取操作系统的方法 这叫做从用户态切换到内核态

操作系统从磁盘把它读取到内核缓冲区

操作系统再从内核缓冲区把数据再赋值到用户缓冲区

用户缓冲区就是 byte 数组

执行到这里，file.read(buf); 就执行完毕了

（相当于从 内核态回到用户态）

然后是写方法的执行

它不能直接把 byte 数组写到网卡中去

首先要把数据写到 socket 缓冲区 ，最后由缓冲区写到网卡

执行读写，一共执行了四次

• 磁盘到内核缓冲区
• 内核缓冲区到用户缓冲区
• 用户缓冲区到 socket缓冲区
• socket缓冲区到网卡

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

优化上面的代码

第一点：DirectByteBuffer 优化

通过 DirectByteBuf

• ByteBuffer.allocate(10) HeapByteBuffer 使用的还是 java 内存
• ByteBuffer.allocateDirect(10) DirectByteBuffer 使用的是操作系统内存

使用了 DirectByteBuffer 内核缓冲区和用户缓冲区是同一块内存，将来java获取的时候，获取的数据就是一模一样的，就减少了一次数据的拷贝

第二点：（linux 2.1）

（底层采用了 linux 2.1 后提供的 sendFile 方法），java 中对应着两个 channel 调用 transferTo/transferFrom 方法拷贝数据

在 fileChancel 方法中才有，底层对应 linux sendFile 方法

之前用 byte 数组复制数据了

现在直接就可以从操作的内核缓冲区把数据复制到socket缓冲区

减少了两次java跟操作系统的切换，而复制数据还是三次

第三点：进一步优化（linux 2.4）

java 调用 transferTo 方法后，要从 java 程序的用户态切换至内核态

数据的复制从磁盘到内核缓冲区

内核缓冲区直接到网卡

要从 java 程序的用户态切换至内核态，使用 DMA将数据读入内核缓冲区，不会使用 cpu（可以解放CPU）

整个过程仅只发生了一次用户态与内核态的切换，数据拷贝了 2 次。所谓的【零拷贝】，并不是真正无拷贝，而是在不会拷贝重复数据到 jvm 内存中，零拷贝的优点有

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

异步例子

异步模型需要底层操作系统（Kernel）提供支持

• Windows 系统通过 IOCP 实现了真正的异步 IO
• Linux 系统异步 IO 在 2.6 版本引入，但其底层实现还是用多路复用模拟了异步 IO，性能没有优势

import lombok.extern.slf4j.Slf4j;

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

import static utils.ByteBufferUtil.debugAll;

@Slf4j
public class fileNett {
    public static void main(String[] args) throws IOException {
        // 参数1：操作文件的路径
        // 参数2：打开文件选项，读，写，追加，覆盖
        // 参数3：线程池
        AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get("data.txt"), StandardOpenOption.READ);
        // 参数1：ByteBuffer
        // 参数2：读取的起始位置
        // 参数3：附件
        // 参数4：回调对象
        ByteBuffer buffer = ByteBuffer.allocate(16);
        log.debug("read begin...");
        channel.read(buffer, 0, buffer, new CompletionHandler<Integer, ByteBuffer>() {
            // 下面的两个方法都是线程，不是主线程

            /**
             *
             * @param result 读取到的字节数
             * @param attachment 附件
             */
            @Override // 读取成功了
            public void completed(Integer result, ByteBuffer attachment) {
                log.debug("read completed...");
                attachment.flip();
                debugAll(attachment);
            }

            /**
             *
             * @param exc
             * @param attachment
             */
            @Override // 读取失败了
            public void failed(Throwable exc, ByteBuffer attachment) {
                exc.printStackTrace();
            }
        });
        log.debug("read end...");
        // 这里卡住，要不然主线程结束了其他线程也结束了
        System.in.read();
    }
}