title: NIO 基礎(chǔ)
date: 2021/04/01 11:16

NIO 基礎(chǔ)

non-blocking io 非阻塞 IO纷纫。

一捷绑、三大組件

1.1 Channel & Buffer

channel 有一點類似于 stream（但 stream 只能單向的讀或者寫）寞射，它就是讀寫數(shù)據(jù)的雙向通道厌蔽，可以從 channel 將數(shù)據(jù)讀入 buffer密末，也可以將 buffer 的數(shù)據(jù)寫入 channel蔼紧，而之前的 stream 要么是輸入婆硬，要么是輸出，channel 比 stream 更為底層

graph LR
channel --> buffer
buffer --> channel

常見的 Channel 有

• FileChannel
• DatagramChannel
• SocketChannel
• ServerSocketChannel

buffer 則用來緩沖讀寫數(shù)據(jù)奸例，常見的 buffer 有

• ByteBuffer
  • MappedByteBuffer
  • DirectByteBuffer
  • HeapByteBuffer
• ShortBuffer
• IntBuffer
• LongBuffer
• FloatBuffer
• DoubleBuffer
• CharBuffer

1.2 Selector

selector 單從字面意思不好理解彬犯，需要結(jié)合服務(wù)器的設(shè)計演化來理解它的用途

多線程版設(shè)計

graph TD
subgraph 多線程版
t1(thread) --> s1(socket1)
t2(thread) --> s2(socket2)
t3(thread) --> s3(socket3)
end

?? 多線程版缺點

• 內(nèi)存占用高
• 線程上下文切換成本高
• 只適合連接數(shù)少的場景

線程池版設(shè)計

graph TD
subgraph 線程池版
t4(thread) --> s4(socket1)
t5(thread) --> s5(socket2)
t4(thread) -.-> s6(socket3)
t5(thread) -.-> s7(socket4)
end

?? 線程池版缺點

• 阻塞模式下，線程僅能處理一個 socket 連接
• 僅適合短連接場景

selector 版設(shè)計

selector 的作用就是配合一個線程來管理多個 channel查吊，獲取這些 channel 上發(fā)生的事件谐区，這些 channel 工作在非阻塞模式下，不會讓線程吊死在一個 channel 上逻卖。適合連接數(shù)特別多宋列，但流量低的場景（low traffic）

graph TD
subgraph selector 版
thread --> selector
selector --> c1(channel)
selector --> c2(channel)
selector --> c3(channel)
end

調(diào)用 selector 的 select() 會阻塞直到 channel 發(fā)生了讀寫就緒事件，這些事件發(fā)生评也，select 方法就會返回這些事件交給 thread 來處理

二炼杖、ByteBuffer

有一普通文本文件 data.txt灭返，內(nèi)容為

1234567890abcd

使用 FileChannel 來讀取文件內(nèi)容，獲取 FileChannel 有兩種方式：

1. new RandomAccessFile("helloword/data.txt", "rw").getChannel();

2. new FileInputStream("helloword/data.txt").getChannel();

   問題：這樣獲取的 Channel 可以向里面寫數(shù)據(jù)嗎坤邪？

   不能熙含，會拋出如下異常：

   java.nio.channels.NonWritableChannelException
   at sun.nio.ch.FileChannelImpl.write(FileChannelImpl.java:201)
   at cn.x5456.nio.study.bf.ByteBufferTest.testReadFile(ByteBufferTest.java:24)

public void testReadFile() {
  try (FileChannel channel = new FileInputStream("helloword/data.txt").getChannel()) {
    // 創(chuàng)建時默認是寫模式
    ByteBuffer byteBuffer = ByteBuffer.allocate(10);
    while (true) {
      int len = channel.read(byteBuffer);
      log.debug("讀取到的字節(jié)數(shù)為：「{}」", len);
      if (len == -1) {
        break;
      }
      // 切換到讀模式
      byteBuffer.flip();
      while (byteBuffer.hasRemaining()) {
        log.debug("讀到的字節(jié)為：「{}」", (char) byteBuffer.get());
      }

      // 切換回寫模式
      byteBuffer.flip();
    }
  } catch (IOException e) {
  }
}

2.1 ByteBuffer 正確使用姿勢

1. 向 buffer 寫入數(shù)據(jù)，例如調(diào)用 channel.read(buffer)
2. 調(diào)用 flip() 切換至讀模式
3. 從 buffer 讀取數(shù)據(jù)艇纺，例如調(diào)用 buffer.get()
4. 調(diào)用 clear() 或 compact() 切換至寫模式
5. 重復 1~4 步驟

2.2 ByteBuffer 結(jié)構(gòu)

ByteBuffer 有以下重要屬性

• capacity
• position：指針索引位置
• limit：當前讀寫的限制

一開始

image

寫模式下怎静，position 是寫入位置，limit 等于容量黔衡，下圖表示寫入了 4 個字節(jié)后的狀態(tài)

image

flip 動作發(fā)生后蚓聘，position 切換為讀取位置，limit 切換為讀取限制

image

讀取 4 個字節(jié)后盟劫，狀態(tài)

image

clear 動作發(fā)生后夜牡，狀態(tài)

image

compact 方法，是把未讀完的部分向前壓縮侣签，然后切換至寫模式

image

2.3 ByteBuffer 常見方法

分配空間

可以使用 allocate 方法為 ByteBuffer 分配空間氯材，其它 buffer 類也有該方法

Bytebuffer buf = ByteBuffer.allocate(16);

向 buffer 寫入數(shù)據(jù)

有兩種辦法

• 調(diào)用 channel 的 read 方法
• 調(diào)用 buffer 自己的 put 方法

int readBytes = channel.read(buf);

和

buf.put((byte)127);

從 buffer 讀取數(shù)據(jù)

同樣有兩種辦法

• 調(diào)用 channel 的 write 方法
• 調(diào)用 buffer 自己的 get 方法

int writeBytes = channel.write(buf);

和

byte b = buf.get();

get 方法會讓 position 讀指針向后走，如果想重復讀取數(shù)據(jù)

• 可以調(diào)用 rewind 方法將 position 重新置為 0
• 或者調(diào)用 get(int i) 方法獲取索引 i 的內(nèi)容硝岗，它不會移動讀指針

mark 和 reset

mark 記錄 position 位置，reset 跳轉(zhuǎn)到 mark 標記的位置袋毙。

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

字符串與 ByteBuffer 互轉(zhuǎn)

public void testConvert() {
    // 字符串轉(zhuǎn) ByteBuffer
    // 方法一
    ByteBuffer buffer = ByteBuffer.allocate(16);
    buffer.put("hello".getBytes(StandardCharsets.UTF_8));

    // 方法二型檀，這種方式會自動把 bytebuffer 切換成讀模式
    ByteBuffer buffer2 = StandardCharsets.UTF_8.encode("hello");

    // 方法三，這種方式會自動把 bytebuffer 切換成讀模式
    ByteBuffer buffer3 = ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8));

    // ByteBuffer 轉(zhuǎn)字符串
    CharBuffer charBuffer = StandardCharsets.UTF_8.decode(buffer3);
    System.out.println(charBuffer.toString());

    // 如果不翻轉(zhuǎn)過來則不會有數(shù)據(jù)听盖，因為后面都是空的
    buffer.flip();
    CharBuffer charBuffer2 = StandardCharsets.UTF_8.decode(buffer);
    System.out.println(charBuffer2.toString());
}

注：Buffer 不是線程安全的

2.4 Scattering Reads

分散讀取胀溺，有一個文本文件 3parts.txt

onetwothree

使用如下方式讀取，可以將數(shù)據(jù)填充至多個 buffer

try (RandomAccessFile file = new RandomAccessFile("helloword/3parts.txt", "rw")) {
    FileChannel channel = file.getChannel();
    ByteBuffer a = ByteBuffer.allocate(3);
    ByteBuffer b = ByteBuffer.allocate(3);
    ByteBuffer c = ByteBuffer.allocate(5);
    channel.read(new ByteBuffer[]{a, b, c});
    a.flip();
    b.flip();
    c.flip();
    debug(a);
    debug(b);
    debug(c);
} catch (IOException e) {
    e.printStackTrace();
}

結(jié)果

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 6f 6e 65                                        |one             |
+--------+-------------------------------------------------+----------------+
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 74 77 6f                                        |two             |
+--------+-------------------------------------------------+----------------+
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 74 68 72 65 65                                  |three           |
+--------+-------------------------------------------------+----------------+

2.5 Gathering Writes

使用如下方式寫入皆看，可以將多個 buffer 的數(shù)據(jù)填充至 channel

try (RandomAccessFile file = new RandomAccessFile("helloword/3parts.txt", "rw")) {
    FileChannel channel = file.getChannel();
    ByteBuffer d = ByteBuffer.allocate(4);
    ByteBuffer e = ByteBuffer.allocate(4);
    channel.position(11);

    d.put(new byte[]{'f', 'o', 'u', 'r'});
    e.put(new byte[]{'f', 'i', 'v', 'e'});
    d.flip();
    e.flip();
    debug(d);
    debug(e);
    channel.write(new ByteBuffer[]{d, e});
} catch (IOException e) {
    e.printStackTrace();
}

輸出

         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 66 6f 75 72                                     |four            |
+--------+-------------------------------------------------+----------------+
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 66 69 76 65                                     |five            |
+--------+-------------------------------------------------+----------------+

文件內(nèi)容

onetwothreefourfive

2.6 練習

網(wǎng)絡(luò)上有多條數(shù)據(jù)發(fā)送給服務(wù)端仓坞，數(shù)據(jù)之間使用 \n 進行分隔
但由于某種原因這些數(shù)據(jù)在接收時，被進行了重新組合腰吟，例如原始數(shù)據(jù)有3條為

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

變成了下面的兩個 byteBuffer (黏包无埃，半包)

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

黏包原因：攢一批數(shù)據(jù)再發(fā)送

半包原因：數(shù)據(jù)超出了接收方緩沖區(qū)的大小

現(xiàn)在要求你編寫程序，將錯亂的數(shù)據(jù)恢復成原始的按 \n 分隔的數(shù)據(jù)

public static void main(String[] args) {
    ByteBuffer source = ByteBuffer.allocate(32);
    //                     11            24
    source.put("Hello,world\nI'm zhangsan\nHo".getBytes());
    split(source);

    source.put("w are you?\nhaha!\n".getBytes());
    split(source);
}

private static void split(ByteBuffer source) {
    // 切換為讀模式
    source.flip();

    // 遍歷當前的 bytebuffer毛雇，找到 \n
    for (int i = 0; i < source.limit(); i++) {
        if (source.get(i) == (byte) '\n') {
            // 獲取到字符串長度
            int l = i - source.position() + 1;
            ByteBuffer target = ByteBuffer.allocate(l);
            for (int j = 0; j < l; j++) {
                // 無參的 get 會移動 position
                target.put(source.get());
            }

            target.flip();
            System.out.println(StandardCharsets.UTF_8.decode(target).toString());
        }
    }

    // 把未讀完的部分向前壓縮嫉称，然后切換至寫模式
    source.compact();
}

三、文件編程

3.1 FileChannel

?? FileChannel 工作模式

FileChannel 只能工作在阻塞模式下

只有 Socket 系列的 Channel 才可以用 Selector

獲取

不能直接打開 FileChannel灵疮，必須通過 FileInputStream织阅、FileOutputStream 或者 RandomAccessFile 來獲取 FileChannel，它們都有 getChannel 方法

• 通過 FileInputStream 獲取的 channel 只能讀
• 通過 FileOutputStream 獲取的 channel 只能寫
• 通過 RandomAccessFile 是否能讀寫根據(jù)構(gòu)造 RandomAccessFile 時的讀寫模式?jīng)Q定

讀取

會從 channel 讀取數(shù)據(jù)填充 ByteBuffer震捣，返回值表示讀到了多少字節(jié)荔棉，-1 表示到達了文件的末尾

int readBytes = channel.read(buffer);

寫入

寫入的正確姿勢如下闹炉， SocketChannel

ByteBuffer buffer = ...;
buffer.put(...); // 存入數(shù)據(jù)
buffer.flip();   // 切換讀模式

while(buffer.hasRemaining()) {
    channel.write(buffer);
}

在 while 中調(diào)用 channel.write 是因為 write 方法并不能保證一次將 buffer 中的內(nèi)容全部寫入 channel。

關(guān)閉

channel 必須關(guān)閉润樱，不過調(diào)用了 FileInputStream渣触、FileOutputStream 或者 RandomAccessFile 的 close 方法會間接地調(diào)用 channel 的 close 方法。

位置

獲取當前位置

long pos = channel.position();

設(shè)置當前位置

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

設(shè)置當前位置時祥国，如果設(shè)置為文件的末尾

• 這時讀取會返回 -1
• 這時寫入昵观，會追加內(nèi)容，但要注意如果 position 超過了文件末尾舌稀，再寫入時在新內(nèi)容和原末尾之間會有空洞（00）

大小

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

強制寫入

操作系統(tǒng)出于性能的考慮啊犬，會將數(shù)據(jù)緩存，不是立刻寫入磁盤壁查。可以調(diào)用 force(true) 方法將文件內(nèi)容和元數(shù)據(jù)（文件的權(quán)限等信息）立刻寫入磁盤

3.2 兩個 Channel 傳輸數(shù)據(jù)

String FROM = "helloword/data.txt";
String TO = "helloword/to.txt";
long start = System.nanoTime();
try (FileChannel from = new FileInputStream(FROM).getChannel();
     FileChannel to = new FileOutputStream(TO).getChannel();
    ) {
      // 操作的時候會使用操作系統(tǒng)的 0 拷貝進行優(yōu)化觉至，傳輸上限為 2G
    from.transferTo(0, from.size(), to);
} catch (IOException e) {
    e.printStackTrace();
}
long end = System.nanoTime();
System.out.println("transferTo 用時：" + (end - start) / 1000_000.0);

輸出

transferTo 用時：8.2011

超過 2g 大小的文件傳輸

public class TestFileChannelTransferTo {
    public static void main(String[] args) {
        try (
                FileChannel from = new FileInputStream("data.txt").getChannel();
                FileChannel to = new FileOutputStream("to.txt").getChannel();
        ) {
            // 效率高，底層會利用操作系統(tǒng)的零拷貝進行優(yōu)化
            long size = from.size();
            // left 變量代表還剩余多少字節(jié)
            for (long left = size; left > 0; ) {
                System.out.println("position:" + (size - left) + " left:" + left);
                left -= from.transferTo((size - left), left, to);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

實際傳輸一個超大文件

position:0 left:7769948160
position:2147483647 left:5622464513
position:4294967294 left:3474980866
position:6442450941 left:1327497219

3.3 Path

jdk7 引入了 Path 和 Paths 類

• Path 用來表示文件路徑
• Paths 是工具類睡腿，用來獲取 Path 實例

Path source = Paths.get("1.txt"); // 相對路徑 使用 user.dir 環(huán)境變量來定位 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

• . 代表了當前路徑
• .. 代表了上一級路徑

例如目錄結(jié)構(gòu)如下

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

代碼

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

檢查文件是否存在

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

創(chuàng)建一級目錄

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

• 如果目錄已存在，會拋異常 FileAlreadyExistsException
• 不能一次創(chuàng)建多級目錄席怪，否則會拋異常 NoSuchFileException

創(chuàng)建多級目錄用

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);

• 如果目錄還有內(nèi)容碉纺，會拋異常 DirectoryNotEmptyException

遍歷文件夾

遍歷目錄文件（訪問者模式）

public static void main(String[] args) throws IOException {
    Path path = Paths.get("C:\\Program Files\\Java\\jdk1.8.0_91");
    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); // 133
    System.out.println(fileCount); // 1479
}

刪除多級目錄

Path path = Paths.get("d:\\a");
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);
    }
});

?? 刪除很危險

刪除是危險操作，確保要遞歸刪除的文件夾沒有重要內(nèi)容

拷貝多級目錄

long start = System.currentTimeMillis();
String source = "D:\\Snipaste-1.16.2-x64";
String target = "D:\\Snipaste-1.16.2-x64aaa";

Files.walk(Paths.get(source)).forEach(path -> {
    try {
        String targetName = path.toString().replace(source, target);
        // 是目錄
        if (Files.isDirectory(path)) {
            Files.createDirectory(Paths.get(targetName));
        }
        // 是普通文件
        else if (Files.isRegularFile(path)) {
            Files.copy(path, Paths.get(targetName));
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
});
long end = System.currentTimeMillis();
System.out.println(end - start);

四刻撒、網(wǎng)絡(luò)編程

4.1 非阻塞 vs 阻塞

阻塞

• 阻塞模式下骨田，相關(guān)方法都會導致線程暫停（最大的弊端，與單線程結(jié)合的不好）
  • ServerSocketChannel.accept 會在沒有連接建立時讓線程暫停
  • SocketChannel.read 會在沒有數(shù)據(jù)可讀時讓線程暫停
  • 阻塞的表現(xiàn)其實就是線程暫停了声怔，暫停期間不會占用 cpu态贤，但線程相當于閑置
• 單線程下，阻塞方法之間相互影響醋火，幾乎不能正常工作悠汽，需要多線程支持
• 但多線程下，有新的問題芥驳，體現(xiàn)在以下方面
  • 32 位 jvm 一個線程 320k介粘，64 位 jvm 一個線程 1024k，如果連接數(shù)過多晚树，必然導致 OOM姻采，并且線程太多，反而會因為頻繁上下文切換導致性能降低
  • 可以采用線程池技術(shù)來減少線程數(shù)和線程上下文切換，但治標不治本慨亲，如果有很多連接建立婚瓜，但長時間 inactive，會阻塞線程池中所有線程刑棵，因此不適合長連接巴刻，只適合短連接。

服務(wù)端

// 使用 nio 來理解阻塞模式, 單線程
// 0. ByteBuffer
ByteBuffer buffer = ByteBuffer.allocate(16);
// 1. 創(chuàng)建了服務(wù)器
ServerSocketChannel ssc = ServerSocketChannel.open();

// 2. 綁定監(jiān)聽端口
ssc.bind(new InetSocketAddress(8080));

// 3. 連接集合
List<SocketChannel> channels = new ArrayList<>();
while (true) {
    // 4. accept 建立與客戶端連接蛉签， SocketChannel 用來與客戶端之間通信
    log.debug("connecting...");
    SocketChannel sc = ssc.accept(); // 阻塞方法胡陪，線程停止運行
    log.debug("connected... {}", sc);
    channels.add(sc);
    for (SocketChannel channel : channels) {
        // 5. 接收客戶端發(fā)送的數(shù)據(jù)
        log.debug("before read... {}", channel);
        channel.read(buffer); // 阻塞方法，線程停止運行
        buffer.flip();
        debugRead(buffer);
        buffer.clear();
        log.debug("after read...{}", channel);
    }
}

客戶端

SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress("localhost", 8080));
System.out.println("waiting...");

非阻塞

• 非阻塞模式下碍舍，相關(guān)方法都會不會讓線程暫停
  • 在 ServerSocketChannel.accept 在沒有連接建立時柠座，會返回 null手报，繼續(xù)運行
  • SocketChannel.read 在沒有數(shù)據(jù)可讀時浩销，會返回 0，但線程不必阻塞宇挫，可以去執(zhí)行其它 SocketChannel 的 read 或是去執(zhí)行 ServerSocketChannel.accept
  • 寫數(shù)據(jù)時捧书，線程只是等待數(shù)據(jù)寫入 Channel 即可吹泡，無需等 Channel 通過網(wǎng)絡(luò)把數(shù)據(jù)發(fā)送出去
• 但非阻塞模式下，即使沒有連接建立经瓷，和可讀數(shù)據(jù)爆哑，線程仍然在不斷運行，白白浪費了 cpu
• 數(shù)據(jù)復制過程中舆吮，線程實際還是阻塞的（AIO 改進的地方）==>看下面的 IO 模型講解

服務(wù)器端泪漂，客戶端代碼不變

// 使用 nio 來理解非阻塞模式, 單線程
// 0. ByteBuffer
ByteBuffer buffer = ByteBuffer.allocate(16);
// 1. 創(chuàng)建了服務(wù)器
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false); // 非阻塞模式，影響的是 accept 方法
// 2. 綁定監(jiān)聽端口
ssc.bind(new InetSocketAddress(8080));
// 3. 連接集合
List<SocketChannel> channels = new ArrayList<>();
while (true) {
    // 4. accept 建立與客戶端連接歪泳， SocketChannel 用來與客戶端之間通信
    SocketChannel sc = ssc.accept(); // 非阻塞，線程還會繼續(xù)運行露筒，如果沒有連接建立呐伞，但 sc 是null
    if (sc != null) {
        log.debug("connected... {}", sc);
        sc.configureBlocking(false); // 非阻塞模式，影響的是 read 方法
        channels.add(sc);
    }
    for (SocketChannel channel : channels) {
        // 5. 接收客戶端發(fā)送的數(shù)據(jù)
        int read = channel.read(buffer);// 非阻塞慎式，線程仍然會繼續(xù)運行伶氢，如果沒有讀到數(shù)據(jù)，read 返回 0
        if (read > 0) {
            buffer.flip();
            debugRead(buffer);
            buffer.clear();
            log.debug("after read...{}", channel);
        }
    }
}

多路復用

單線程可以配合 Selector 完成對多個 Channel 可讀寫事件的監(jiān)控瘪吏，這稱之為多路復用

• 多路復用僅針對網(wǎng)絡(luò) IO癣防、普通文件 IO 沒法利用多路復用
• 如果不用 Selector 的非阻塞模式，線程大部分時間都在做無用功掌眠，而 Selector 能夠保證
  • 有可連接事件時才去連接
  • 有可讀事件才去讀取
  • 有可寫事件才去寫入
    • 限于網(wǎng)絡(luò)傳輸能力蕾盯，Channel 未必時時可寫，一旦 Channel 可寫蓝丙，會觸發(fā) Selector 的可寫事件

4.2 Selector

graph TD
subgraph selector 版
thread --> selector
selector --> c1(channel)
selector --> c2(channel)
selector --> c3(channel)
end

好處

• 一個線程配合 selector 就可以監(jiān)控多個 channel 的事件级遭，事件發(fā)生線程才去處理望拖。避免非阻塞模式下所做無用功
• 讓這個線程能夠被充分利用
• 節(jié)約了線程的數(shù)量
• 減少了線程上下文切換

創(chuàng)建

Selector selector = Selector.open();

綁定 Channel 事件

也稱之為注冊事件，綁定的事件 selector 才會關(guān)心

channel.configureBlocking(false);
SelectionKey key = channel.register(selector, 綁定事件);

• channel 必須工作在非阻塞模式
• FileChannel 沒有非阻塞模式挫鸽，因此不能配合 selector 一起使用
• 綁定的事件類型可以有
  • connect - 客戶端連接成功時觸發(fā)
  • accept - 服務(wù)器端成功接受連接時觸發(fā)
  • read - 數(shù)據(jù)可讀入時觸發(fā)说敏，有因為接收能力弱，數(shù)據(jù)暫不能讀入的情況
  • write - 數(shù)據(jù)可寫出時觸發(fā)丢郊，有因為發(fā)送能力弱盔沫，數(shù)據(jù)暫不能寫出的情況

監(jiān)聽 Channel 事件

可以通過下面三種方法來監(jiān)聽是否有事件發(fā)生，方法的返回值代表有多少 channel 發(fā)生了事件

方法1枫匾，阻塞直到綁定事件發(fā)生

int count = selector.select();

方法2架诞，阻塞直到綁定事件發(fā)生，或是超時（時間單位為 ms）

int count = selector.select(long timeout);

方法3婿牍，不會阻塞侈贷，也就是不管有沒有事件，立刻返回等脂，自己根據(jù)返回值檢查是否有事件

int count = selector.selectNow();

?? select 何時不阻塞

• 事件發(fā)生時
  • 客戶端發(fā)起連接請求俏蛮，會觸發(fā) accept 事件
  • 客戶端發(fā)送數(shù)據(jù)過來，客戶端正常上遥、異常關(guān)閉時搏屑，都會觸發(fā) read 事件，另外如果發(fā)送的數(shù)據(jù)大于 buffer 緩沖區(qū)粉楚，會觸發(fā)多次讀取事件
  • channel 可寫辣恋，會觸發(fā) write 事件
  • 在 linux 下 nio bug 發(fā)生時
• 調(diào)用 selector.wakeup()
• 調(diào)用 selector.close()
• selector 所在線程 interrupt

4.3 Selector 實現(xiàn)服務(wù)端

// 服務(wù)器端
public static void main(String[] args) {
    try (ServerSocketChannel ssc = ServerSocketChannel.open();
         Selector selector = Selector.open()) {

        // 監(jiān)聽 8080 端口
        ssc.bind(new InetSocketAddress(8080));
        log.info("服務(wù)器 channel：{}", ssc);

        // 設(shè)置服務(wù)器 channel 為非阻塞
        ssc.configureBlocking(false);

        // 將服務(wù)器 channel 注冊到 selector 中，交給 selector 監(jiān)聽 accept 事件
        SelectionKey ssk = ssc.register(selector, SelectionKey.OP_ACCEPT);
        log.info("服務(wù)器的 SelectionKey：{}", ssk);

        while (true) {
            // 阻塞方法模软，只有有事件的時候才會返回
            int count = selector.select();

            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();

                // 處理key 時伟骨，要從 selectedKeys 集合中刪除，否則下次處理的時候還會拿到這個 SelectionKey
                iterator.remove();

                // 如果是『接受連接』事件
                if (key.isAcceptable()) {
                    log.info("這個地方的 selectKey 和上面的服務(wù)器的 SelectionKey 一毛一樣：{}", key);
                    ServerSocketChannel channel = (ServerSocketChannel) key.channel();

                    log.info("如果這個地方不對事件進行處理（即把下面代碼注釋掉）燃异，下次該事件仍會觸發(fā)携狭，這是因為 nio 底層使用的是水平觸發(fā)。");

                    // 與客戶端建立連接回俐，返回客戶端的 channel
                    SocketChannel sc = channel.accept();
                    sc.configureBlocking(false);
                    log.info("客戶端 Channel：{}", sc);

                    // 訂閱客戶端 channel 上的 read 事件
                    SelectionKey scKey = sc.register(selector, SelectionKey.OP_READ);
                    log.info("客戶端的 SelectionKey：{}", scKey);
                } else if (key.isReadable()) {  // 如果是『讀』事件逛腿，那么肯定是客戶端 channel 觸發(fā)的，所以可以把它強轉(zhuǎn)成 SocketChannel
                    log.info("這個地方的 selectKey 和上面建立連接時獲取到的客戶端 SelectionKey 一毛一樣：{}", key);
                    try {
                        // 拿到觸發(fā)事件的channel
                        SocketChannel channel = (SocketChannel) key.channel();
                        // 如果客戶端發(fā)來的消息超過 4 個字節(jié)仅颇，則會觸發(fā)多次 read 事件单默，因為 nio 底層使用的是水平觸發(fā)。
                        ByteBuffer buffer = ByteBuffer.allocate(4);
                        // 讀取 channel 中的字節(jié)到 buffer 中
                        int read = channel.read(buffer);
                        // 如果是正常斷開忘瓦，也會觸發(fā)一次 read 事件搁廓，此時 read 的方法的返回值是 -1
                        if (read == -1) {
                            log.info("客戶端「{}」關(guān)閉了", channel);
                            // 必須對這個鍵進行處理，否則會一直觸發(fā)
                            key.cancel();
                        } else {
                            /*
                            +--------+-------------------- all ------------------------+----------------+
                            position: [4], limit: [4]
                                     +-------------------------------------------------+
                                     |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
                            +--------+-------------------------------------------------+----------------+
                            |00000000| 61 62 63 64                                     |abcd            |
                            +--------+-------------------------------------------------+----------------+
                            10:14:04.838 [main] INFO cn.x5456.nio.study.sk.SelectorServer - 這個地方的 selectKey 和上面建立連接時獲取到的客戶端 SelectionKey 一毛一樣：sun.nio.ch.SelectionKeyImpl@ba8a1dc
                            +--------+-------------------- all ------------------------+----------------+
                            position: [2], limit: [4]
                                     +-------------------------------------------------+
                                     |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
                            +--------+-------------------------------------------------+----------------+
                            |00000000| 65 66 00 00                                     |ef..            |
                            +--------+-------------------------------------------------+----------------+
                             */
                            // 打印讀取到的內(nèi)容
                            debugAll(buffer);
                        }
                    } catch (IOException e) {
                        // 當客戶端異常斷開會拋出異常，我們需要取消對這個 key 的監(jiān)聽
                        e.printStackTrace();
                        key.cancel();
                    }
                }
            }
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}

// 客戶端
public static void main(String[] args) throws IOException {
    SocketChannel sc = SocketChannel.open();
    sc.connect(new InetSocketAddress("localhost", 8080));
    sc.write(ByteBuffer.wrap("abcdef".getBytes(StandardCharsets.UTF_8)));
}

?? 事件發(fā)生后能否不處理

事件發(fā)生后枚抵，要么處理线欲，要么取消（cancel），不能什么都不做汽摹，否則下次該事件仍會觸發(fā)李丰，這是因為 nio 底層使用的是水平觸發(fā)。

根本原因：server沒有讀取socketChannel中的數(shù)據(jù)

水平觸發(fā)(level-triggered逼泣，也被稱為條件觸發(fā))LT: 只要滿足條件趴泌，就觸發(fā)一個事件(只要有數(shù)據(jù)沒有被獲取，內(nèi)核就不斷通知你)

邊緣觸發(fā)(edge-triggered)ET: 每當狀態(tài)變化時拉庶，觸發(fā)一個事件嗜憔。

?? 為何要 iter.remove()

因為 select 在事件發(fā)生后，就會將相關(guān)的 key 放入 selectedKeys 集合氏仗，但不會在處理完后從 selectedKeys 集合中移除吉捶，需要我們自己編碼刪除。例如

• 第一次觸發(fā)了 ssckey 上的 accept 事件皆尔，沒有移除 ssckey
• 第二次觸發(fā)了 sckey 上的 read 事件呐舔，但這時 selectedKeys 中還有上次的 ssckey ，在處理時因為沒有真正的 serverSocket 連上了慷蠕，就會導致空指針異常[圖片上傳失敗...(image-a43c9b-1617350937261)]

?? cancel 的作用

cancel 會取消注冊在 selector 上的 channel珊拼，并從 keys 集合中刪除 key 后續(xù)不會再監(jiān)聽事件。

??如何處理消息的邊界

image

1. 一種思路是固定消息長度流炕，數(shù)據(jù)包大小一樣澎现，服務(wù)器按預(yù)定長度讀取，缺點是浪費帶寬（因為有些消息小于這個固定長度）

2. 另一種思路是按分隔符拆分每辟，缺點是效率低（因為需要一個一個字節(jié)對比剑辫；上面的練習）

3. TLV 格式，即 Type 類型渠欺、Length 長度妹蔽、Value 數(shù)據(jù)，類型和長度已知的情況下峻堰，就可以方便獲取消息大小，分配合適的 buffer盅视，缺點是 buffer 需要提前分配捐名，如果內(nèi)容過大，則影響 server 吞吐量（Netty 使用這種方式）

   1. Http 1.1 是 TLV 格式（Content-Type闹击、Content-Length镶蹋、Body）
   2. Http 2.0 是 LTV 格式

本示例采用第二種方式進行解決：

sequenceDiagram 
participant c1 as 客戶端1
participant s as 服務(wù)器
participant b1 as ByteBuffer1
participant b2 as ByteBuffer2
c1 ->> s: 發(fā)送 01234567890abcdef3333\r
s ->> b1: 第一次 read 存入 01234567890abcdef
s ->> b2: 擴容
b1 ->> b2: 拷貝 01234567890abcdef
s ->> b2: 第二次 read 存入 3333\r
b2 ->> b2: 01234567890abcdef3333\r

// 服務(wù)器端
@Slf4j
public class SelectorServer {

    public static void main(String[] args) {
        try (ServerSocketChannel ssc = ServerSocketChannel.open();
             Selector selector = Selector.open()) {

            // 監(jiān)聽 8080 端口
            ssc.bind(new InetSocketAddress(8080));
            log.info("服務(wù)器 channel：{}", ssc);

            // 設(shè)置服務(wù)器 channel 為非阻塞
            ssc.configureBlocking(false);

            // 將服務(wù)器 channel 注冊到 selector 中，交給 selector 監(jiān)聽 accept 事件
            SelectionKey ssk = ssc.register(selector, SelectionKey.OP_ACCEPT);
            log.info("服務(wù)器的 SelectionKey：{}", ssk);

            while (true) {
                // 阻塞方法，只有有事件的時候才會返回
                int count = selector.select();

                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();

                    // 處理key 時贺归，要從 selectedKeys 集合中刪除淆两，否則下次處理的時候還會拿到這個 SelectionKey
                    iterator.remove();

                    // 如果是『接受連接』事件
                    if (key.isAcceptable()) {
                        log.info("這個地方的 selectKey 和上面的服務(wù)器的 SelectionKey 一毛一樣：{}", key);
                        ServerSocketChannel channel = (ServerSocketChannel) key.channel();

                        log.info("如果這個地方不對事件進行處理（即把下面代碼注釋掉），下次該事件仍會觸發(fā)拂酣，這是因為 nio 底層使用的是水平觸發(fā)秋冰。");

                        // 與客戶端建立連接，返回客戶端的 channel
                        SocketChannel sc = channel.accept();
                        sc.configureBlocking(false);
                        log.info("客戶端 Channel：{}", sc);

                        // 訂閱客戶端 channel 上的 read 事件
                        // 【解決黏包問題婶熬，步驟 0】剑勾，添加附件
                        SelectionKey scKey = sc.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(4));
                        log.info("客戶端的 SelectionKey：{}", scKey);
                    } else if (key.isReadable()) {  // 如果是『讀』事件，那么肯定是客戶端 channel 觸發(fā)的赵颅，所以可以把它強轉(zhuǎn)成 SocketChannel
                        log.info("這個地方的 selectKey 和上面建立連接時獲取到的客戶端 SelectionKey 一毛一樣：{}", key);
                        try {
                            // 拿到觸發(fā)事件的channel
                            SocketChannel channel = (SocketChannel) key.channel();
                            // 【解決黏包問題虽另，步驟 1】，從附件中獲取 channel 對應(yīng)的 buffer（不能用全局的饺谬，因為多個客戶端的數(shù)據(jù)會混合）
                            ByteBuffer buffer = (ByteBuffer) key.attachment();
                            // 讀取 channel 中的字節(jié)到 buffer 中
                            int read = channel.read(buffer);
                            // 如果是正常斷開捂刺，也會觸發(fā)一次 read 事件，此時 read 的方法的返回值是 -1
                            if (read == -1) {
                                log.info("客戶端「{}」關(guān)閉了", channel);
                                // 必須對這個鍵進行處理募寨，否則會一直觸發(fā)
                                key.cancel();
                            } else {
                                // 【解決黏包問題族展，步驟 2】
                                split(buffer);
                                // 『解決半包問題』，擴容绪商；如果不擴容會陷入死循環(huán)苛谷，因為 channel 中的數(shù)據(jù)一直沒有讀完
                                if (buffer.limit() == buffer.position()) {
                                    ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
                                    // 將 src buffer 切換為讀模式，因為 split(buffer) 給他切換成了寫模式
                                    buffer.flip();
                                    newBuffer.put(buffer);
                                    key.attach(newBuffer);
                                }
                            }
                        } catch (IOException e) {
                            // 當客戶端異常斷開會拋出異常格郁，我們需要取消對這個 key 的監(jiān)聽
                            e.printStackTrace();
                            key.cancel();
                        }
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void split(ByteBuffer source) {
        // 切換為讀模式
        source.flip();

        // 遍歷當前的 bytebuffer腹殿，找到 \n
        for (int i = 0; i < source.limit(); i++) {
            if (source.get(i) == (byte) '\n') {
                // 獲取到字符串長度
                int l = i - source.position() + 1;
                ByteBuffer target = ByteBuffer.allocate(l);
                for (int j = 0; j < l; j++) {
                    // 無參的 get 會移動 position
                    target.put(source.get());
                }

                target.flip();
                debugAll(target);
            }
        }

        // 把未讀完的部分向前壓縮，然后切換至寫模式
        source.compact();
    }
}

// 客戶端
public class Client {
    public static void main(String[] args) throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 8080));
        // 解決黏包問題
//        sc.write(ByteBuffer.wrap("ab\ncde\nf\n".getBytes(StandardCharsets.UTF_8)));
        // 解決半包問題
        sc.write(ByteBuffer.wrap("1234567890\n".getBytes(StandardCharsets.UTF_8)));
    }
}

==附件這個感覺可以學下啊==

ByteBuffer 大小分配

• 每個 channel 都需要記錄可能被切分的消息例书，因為 ByteBuffer 不能被多個 channel 共同使用锣尉，因此需要為每個 channel 維護一個獨立的 ByteBuffer
• ByteBuffer 不能太大，比如一個 ByteBuffer 1Mb 的話决采，要支持百萬連接就要 1Tb 內(nèi)存自沧，因此需要設(shè)計大小可變的 ByteBuffer
  • 一種思路是首先分配一個較小的 buffer，例如 4k树瞭，如果發(fā)現(xiàn)數(shù)據(jù)不夠拇厢，再分配 8k 的 buffer，將 4k buffer 內(nèi)容拷貝至 8k buffer晒喷，優(yōu)點是消息連續(xù)容易處理孝偎，缺點是數(shù)據(jù)拷貝耗費性能，參考實現(xiàn) http://tutorials.jenkov.com/java-performance/resizable-array.html
  • 另一種思路是用多個數(shù)組組成 buffer凉敲，一個數(shù)組不夠衣盾，把多出來的內(nèi)容寫入新的數(shù)組寺旺，與前面的區(qū)別是消息存儲不連續(xù)解析復雜，優(yōu)點是避免了拷貝引起的性能損耗（Netty 使用的這種方式）

4.4 處理 write 事件

非阻塞模式下势决，無法保證把 buffer 中所有數(shù)據(jù)都寫入 channel阻塑，因此需要追蹤 write 方法的返回值（代表實際寫入字節(jié)數(shù)）:

public static void main(String[] args) throws IOException {
    Selector selector = Selector.open();

    ServerSocketChannel ssc = ServerSocketChannel.open();
    ssc.bind(new InetSocketAddress(8080));
    ssc.configureBlocking(false);
    ssc.register(selector, SelectionKey.OP_ACCEPT);

    while (true) {
        int count = selector.select();
        Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
        while (iter.hasNext()) {
            SelectionKey selectionKey = iter.next();
            iter.remove();

            if (selectionKey.isAcceptable()) {
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);

                // 向客戶端發(fā)送大量數(shù)據(jù)
                StringBuilder sb = new StringBuilder();
                for (int i = 0; i < 5000000; i++) {
                    sb.append("a");
                }

                ByteBuffer buffer = StandardCharsets.UTF_8.encode(sb.toString());
                while (buffer.hasRemaining()) {
                    int write = sc.write(buffer);
                    log.info("寫入的字節(jié)數(shù)：{}", write);
                }
            }
        }
    }
}

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

    // 3. 接收數(shù)據(jù)
    int count = 0;
    while (true) {
        ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
        count += sc.read(buffer);
        System.out.println(count);
        buffer.clear();
    }
}

image

所以我們應(yīng)該改為使用 selector 來監(jiān)聽可寫事件，但是用 selector 監(jiān)聽所有 channel 的可寫事件果复，每個 channel 都需要一個 key 來跟蹤 buffer陈莽，但這樣又會導致占用內(nèi)存過多，就有兩階段策略：

• 當消息處理器第一次寫入消息時据悔，才將 channel 注冊到 selector 上
• selector 檢查 channel 上的可寫事件传透，如果所有的數(shù)據(jù)寫完了，就取消 channel 的注冊
• 如果不取消极颓，會每次可寫均會觸發(fā) write 事件

public static void main(String[] args) throws IOException {
    Selector selector = Selector.open();

    ServerSocketChannel ssc = ServerSocketChannel.open();
    ssc.bind(new InetSocketAddress(8080));
    ssc.configureBlocking(false);
    ssc.register(selector, SelectionKey.OP_ACCEPT);

    while (true) {
        int count = selector.select();
        Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
        while (iter.hasNext()) {
            SelectionKey selectionKey = iter.next();
            iter.remove();

            if (selectionKey.isAcceptable()) {
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                SelectionKey sk = sc.register(selector, SelectionKey.OP_READ);

                // 1. 向客戶端發(fā)送大量數(shù)據(jù)
                StringBuilder sb = new StringBuilder();
                for (int i = 0; i < 5000000; i++) {
                    sb.append("a");
                }

                ByteBuffer buffer = StandardCharsets.UTF_8.encode(sb.toString());
                int write = sc.write(buffer);
                log.info("寫入的字節(jié)數(shù)：{}", write);

                // 2. 如果還有剩余的內(nèi)容朱盐，則關(guān)注可寫事件
                if (buffer.hasRemaining()) {
                    // 使用 | 運算符也行
                    sk.interestOps(sk.interestOps() + SelectionKey.OP_WRITE);
                    // 把未寫完的 buffer 綁定到 sk 上
                    sk.attach(buffer);
                }
            } else if (selectionKey.isWritable()) {
                ByteBuffer buffer = (ByteBuffer) selectionKey.attachment();
                SocketChannel sc = (SocketChannel) selectionKey.channel();
                int write = sc.write(buffer);
                log.info("寫入的字節(jié)數(shù)：{}", write);
                // 3. 如果寫完了，則進行清理操作
                if (!buffer.hasRemaining()) {
                    selectionKey.attach(null);
                    selectionKey.interestOps(selectionKey.interestOps() - SelectionKey.OP_WRITE);
                }
            }
        }
    }
}

?? write 為何要取消

只要向 channel 發(fā)送數(shù)據(jù)時菠隆，socket 緩沖可寫兵琳，這個事件會頻繁觸發(fā)，因此應(yīng)當只在 socket 緩沖區(qū)寫不下時再關(guān)注可寫事件骇径，數(shù)據(jù)寫完之后再取消關(guān)注

4.5 使用多線程優(yōu)化

現(xiàn)在都是多核 cpu躯肌，設(shè)計時要充分考慮別讓 cpu 的力量被白白浪費

前面的代碼只有一個選擇器，沒有充分利用多核 cpu破衔，如何改進呢清女？

分兩組選擇器

• 單線程配一個選擇器，專門處理 accept 事件
• 創(chuàng)建 cpu 核心數(shù)的線程晰筛，每個線程配一個選擇器嫡丙，輪流處理 read 事件

@Slf4j
public class MultiThreadServer {

    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss");

        Selector boss = Selector.open();

        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.bind(new InetSocketAddress(8080));
        ssc.configureBlocking(false);
        SelectionKey bossKey = ssc.register(boss, SelectionKey.OP_ACCEPT);

        // 這個可以建立多個 worker
        Worker worker = new Worker("worker");

        while (true) {
            int count = boss.select();
            Iterator<SelectionKey> iterator = boss.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();

                if (key.isAcceptable()) {
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);

                    // 關(guān)聯(lián)到 work 的 selector
                    log.debug("before register...{}", sc.getRemoteAddress());
                    worker.register(sc);
                    log.debug("after register...{}", sc.getRemoteAddress());
                }
            }
        }
    }


    static class Worker implements Runnable {
        private String name;
        private Selector selector;
        private Thread thread;

        private ConcurrentLinkedQueue<Runnable> tasks = new ConcurrentLinkedQueue<>();

        public Worker(String name) throws IOException {
            this.name = name;
            this.selector = Selector.open();
            this.thread = new Thread(this, name);
            this.thread.start();
        }

        // 這個方法的操作是在 boss 線程中的
        public void register(SocketChannel sc) throws ClosedChannelException {
            tasks.add(() -> {
                try {
                    sc.register(selector, SelectionKey.OP_READ);
                } catch (ClosedChannelException e) {
                    e.printStackTrace();
                }
            });

            // 由于該 selector 已經(jīng)在 worker 線程中被阻塞了，所以只有喚醒它才能進行注冊的操作
            selector.wakeup();
        }

        @Override
        public void run() {
            while (true) {
                try {
                    int count = selector.select();
                    Runnable task = tasks.poll();
                    if (task != null) {
                        task.run();
                    }

                    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("read...{}", channel.getRemoteAddress());
                            channel.read(buffer);
                            buffer.flip();
                            debugAll(buffer);
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}

5. NIO vs BIO

5.1 stream vs channel

• stream 不會自動緩沖數(shù)據(jù)（byte[]不屬于系統(tǒng)層面）读第，channel 會利用系統(tǒng)提供的發(fā)送緩沖區(qū)曙博、接收緩沖區(qū)（更為底層）
• stream 僅支持阻塞 API，channel 同時支持阻塞怜瞒、非阻塞 API父泳，網(wǎng)絡(luò) channel 可配合 selector 實現(xiàn)多路復用
• 二者均為全雙工，即讀寫可以同時進行（可以對一個文件一邊讀一邊寫）

5.2 IO 模型

同步阻塞吴汪、同步非阻塞惠窄、同步多路復用、異步阻塞（沒有此情況）漾橙、異步非阻塞

• 同步：線程自己去獲取結(jié)果（一個線程）
• 異步：線程自己不去獲取結(jié)果杆融，而是由其它線程送結(jié)果（至少兩個線程）

當調(diào)用一次 channel.read 或 stream.read 后，會切換至操作系統(tǒng)內(nèi)核態(tài)來完成真正數(shù)據(jù)讀取近刘，而讀取又分為兩個階段擒贸，分別為：

1. 等待數(shù)據(jù)階段
2. 復制數(shù)據(jù)階段

• 阻塞 IO（同步阻塞，阻塞指的是用戶線程被阻塞觉渴，對應(yīng)我們上面的阻塞例子）

  image

• 非阻塞 IO（同步非阻塞介劫，對應(yīng)我們上面的非阻塞例子）

  image

• 多路復用（同步非阻塞，對應(yīng)我們上面 selector 的例子）

  image

• 信號驅(qū)動

• 異步 IO

  image

• 阻塞 IO vs 多路復用

  阻塞 IO
  多路復用

?? 參考

UNIX 網(wǎng)絡(luò)編程 - 卷 I

5.3 零拷貝

傳統(tǒng) IO 問題

傳統(tǒng)的 IO 將一個文件通過 socket 寫出

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);

內(nèi)部工作流程是這樣的：

image

1. java 本身并不具備 IO 讀寫能力案淋，因此 read 方法調(diào)用后座韵，要從 java 程序的用戶態(tài)切換至內(nèi)核態(tài)，去調(diào)用操作系統(tǒng)（Kernel）的讀能力踢京，將數(shù)據(jù)讀入內(nèi)核緩沖區(qū)誉碴。這期間用戶線程阻塞，操作系統(tǒng)使用 DMA（Direct Memory Access）來實現(xiàn)文件讀瓣距，其間也不會使用 cpu

   DMA 也可以理解為硬件單元黔帕，用來解放 cpu 完成文件 IO

2. 從內(nèi)核態(tài)切換回用戶態(tài)，將數(shù)據(jù)從內(nèi)核緩沖區(qū)讀入用戶緩沖區(qū)（即 byte[] buf）蹈丸，這期間 cpu 會參與拷貝成黄，無法利用 DMA

3. 調(diào)用 write 方法，這時將數(shù)據(jù)從用戶緩沖區(qū)（byte[] buf）寫入 socket 緩沖區(qū)逻杖，cpu 會參與拷貝

4. 接下來要向網(wǎng)卡寫數(shù)據(jù)奋岁，這項能力 java 又不具備，因此又得從用戶態(tài)切換至內(nèi)核態(tài)荸百，調(diào)用操作系統(tǒng)的寫能力闻伶，使用 DMA 將 socket 緩沖區(qū)的數(shù)據(jù)寫入網(wǎng)卡，不會使用 cpu

可以看到中間環(huán)節(jié)較多够话，java 的 IO 實際不是物理設(shè)備級別的讀寫蓝翰，而是緩存的復制，底層的真正讀寫是操作系統(tǒng)來完成的

• 用戶態(tài)與內(nèi)核態(tài)的切換發(fā)生了 3 次更鲁，這個操作比較重量級
  • 切換到內(nèi)核態(tài)霎箍，將數(shù)據(jù)從磁盤 copy 到內(nèi)核緩沖區(qū)
  • 切換到用戶態(tài)，將數(shù)據(jù)從用戶緩沖區(qū) copy 到 socket 緩沖區(qū)
  • 切換到內(nèi)核態(tài)澡为，將數(shù)據(jù)從 socket 緩沖區(qū)發(fā)送到網(wǎng)卡漂坏。
• 數(shù)據(jù)拷貝了共 4 次

NIO 優(yōu)化

1）通過 DirectByteBuf

• ByteBuffer.allocate(10) HeapByteBuffer 使用的還是 java 內(nèi)存
• ByteBuffer.allocateDirect(10) DirectByteBuffer 使用的是操作系統(tǒng)內(nèi)存

image-20210402151147334

大部分步驟與優(yōu)化前相同，不再贅述媒至。唯有一點：java 可以使用 DirectByteBuf 將堆外內(nèi)存映射到 jvm 內(nèi)存中來直接訪問使用

• 這塊內(nèi)存不受 jvm 垃圾回收的影響顶别，因此內(nèi)存地址固定，有助于 IO 讀寫
• java 中的 DirectByteBuf 對象僅維護了此內(nèi)存的虛引用拒啰，內(nèi)存回收分成兩步
  • DirectByteBuf 對象被垃圾回收驯绎，將虛引用加入引用隊列
  • 通過專門線程訪問引用隊列，根據(jù)虛引用釋放堆外內(nèi)存
• 減少了一次數(shù)據(jù)拷貝谋旦，用戶態(tài)與內(nèi)核態(tài)的切換次數(shù)沒有減少（還是三次）

2）進一步優(yōu)化（底層采用了 linux 2.1 后提供的 sendFile 方法）剩失，java 中對應(yīng)著兩個 channel 調(diào)用 transferTo/transferFrom 方法拷貝數(shù)據(jù)（比上面的方式減少了2次 java 代碼（用戶態(tài)）到操作系統(tǒng)（內(nèi)核態(tài)）的切換屈尼，連 ByteBuffer 對象都不創(chuàng)建了）

image

1. java 調(diào)用 transferTo 方法后，要從 java 程序的用戶態(tài)切換至內(nèi)核態(tài)拴孤，使用 DMA將數(shù)據(jù)讀入內(nèi)核緩沖區(qū)脾歧，不會使用 cpu
2. 數(shù)據(jù)從內(nèi)核緩沖區(qū)傳輸?shù)?socket 緩沖區(qū)，cpu 會參與拷貝
3. 最后使用 DMA 將 socket 緩沖區(qū)的數(shù)據(jù)寫入網(wǎng)卡演熟，不會使用 cpu

可以看到

• 只發(fā)生了一次用戶態(tài)與內(nèi)核態(tài)的切換
  • 切換到內(nèi)核態(tài)鞭执，從磁盤將數(shù)據(jù) copy 到內(nèi)核緩沖區(qū)，將內(nèi)核緩沖區(qū)數(shù)據(jù) copy 到 socket 緩沖區(qū)芒粹，最后發(fā)送到網(wǎng)卡兄纺。
• 數(shù)據(jù)拷貝了 3 次

3）進一步優(yōu)化（linux 2.4）

image

1. java 調(diào)用 transferTo 方法后，要從 java 程序的用戶態(tài)切換至內(nèi)核態(tài)化漆，使用 DMA將數(shù)據(jù)讀入內(nèi)核緩沖區(qū)估脆，不會使用 cpu
2. 只會將一些 offset 和 length 信息拷入 socket 緩沖區(qū)，幾乎無消耗
3. 使用 DMA 將 內(nèi)核緩沖區(qū)的數(shù)據(jù)寫入網(wǎng)卡座云，不會使用 cpu

整個過程僅只發(fā)生了一次用戶態(tài)與內(nèi)核態(tài)的切換旁蔼，數(shù)據(jù)拷貝了 2 次。所謂的【零拷貝】疙教，并不是真正無拷貝棺聊，而是在不會拷貝重復數(shù)據(jù)到 jvm 內(nèi)存中，零拷貝的優(yōu)點有：

• 更少的用戶態(tài)與內(nèi)核態(tài)的切換
• 不利用 cpu 計算贞谓，減少 cpu 緩存?zhèn)喂蚕恚ㄒ驗榱憧截悤褂?DMA 進行數(shù)據(jù)的 copy限佩，根本沒有放入內(nèi)存，所以 cpu 無法參與計算）
• 零拷貝適合小文件傳輸（文件較大會把內(nèi)核緩沖區(qū)占滿裸弦，https://www.cnblogs.com/-wenli/p/13380616.html）

5.3 AIO

AIO 用來解決數(shù)據(jù)復制階段的阻塞問題

• 同步意味著祟同，在進行讀寫操作時，線程需要等待結(jié)果理疙，還是相當于閑置
• 異步意味著晕城，在進行讀寫操作時，線程不必等待結(jié)果窖贤，而是將來由操作系統(tǒng)來通過回調(diào)方式由另外的線程來獲得結(jié)果

異步模型需要底層操作系統(tǒng)（Kernel）提供支持

• Windows 系統(tǒng)通過 IOCP 實現(xiàn)了真正的異步 IO
• Linux 系統(tǒng)異步 IO 在 2.6 版本引入砖顷，但其底層實現(xiàn)還是用多路復用模擬了異步 IO，性能沒有優(yōu)勢

文件 AIO

先來看看 AsynchronousFileChannel

@Slf4j
public class AioDemo1 {
    public static void main(String[] args) throws IOException {
        try{
            AsynchronousFileChannel s = 
                AsynchronousFileChannel.open(
                    Paths.get("1.txt"), StandardOpenOption.READ);
            ByteBuffer buffer = ByteBuffer.allocate(2);
            log.debug("begin...");
            s.read(buffer, 0, null, new CompletionHandler<Integer, ByteBuffer>() {
                @Override
                public void completed(Integer result, ByteBuffer attachment) {
                    log.debug("read completed...{}", result);
                    buffer.flip();
                    debug(buffer);
                }

                @Override
                public void failed(Throwable exc, ByteBuffer attachment) {
                    log.debug("read failed...");
                }
            });

        } catch (IOException e) {
            e.printStackTrace();
        }
        log.debug("do other things...");
        System.in.read();
    }
}

輸出

13:44:56 [DEBUG] [main] c.i.aio.AioDemo1 - begin...
13:44:56 [DEBUG] [main] c.i.aio.AioDemo1 - do other things...
13:44:56 [DEBUG] [Thread-5] c.i.aio.AioDemo1 - read completed...2
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 0d                                           |a.              |
+--------+-------------------------------------------------+----------------+

可以看到

• 響應(yīng)文件讀取成功的是另一個線程 Thread-5
• 主線程并沒有 IO 操作阻塞

?? 守護線程

默認文件 AIO 使用的線程都是守護線程赃梧，所以最后要執(zhí)行 System.in.read() 以避免守護線程意外結(jié)束

網(wǎng)絡(luò) AIO

public class AioServer {
    public static void main(String[] args) throws IOException {
        AsynchronousServerSocketChannel ssc = AsynchronousServerSocketChannel.open();
        ssc.bind(new InetSocketAddress(8080));
        ssc.accept(null, new AcceptHandler(ssc));
        System.in.read();
    }

    private static void closeChannel(AsynchronousSocketChannel sc) {
        try {
            System.out.printf("[%s] %s close\n", Thread.currentThread().getName(), sc.getRemoteAddress());
            sc.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {
        private final AsynchronousSocketChannel sc;

        public ReadHandler(AsynchronousSocketChannel sc) {
            this.sc = sc;
        }

        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            try {
                if (result == -1) {
                    closeChannel(sc);
                    return;
                }
                System.out.printf("[%s] %s read\n", Thread.currentThread().getName(), sc.getRemoteAddress());
                attachment.flip();
                System.out.println(Charset.defaultCharset().decode(attachment));
                attachment.clear();
                // 處理完第一個 read 時滤蝠，需要再次調(diào)用 read 方法來處理下一個 read 事件
                sc.read(attachment, attachment, this);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
            closeChannel(sc);
            exc.printStackTrace();
        }
    }

    private static class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {
        private final AsynchronousSocketChannel sc;

        private WriteHandler(AsynchronousSocketChannel sc) {
            this.sc = sc;
        }

        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            // 如果作為附件的 buffer 還有內(nèi)容，需要再次 write 寫出剩余內(nèi)容
            if (attachment.hasRemaining()) {
                sc.write(attachment);
            }
        }

        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
            exc.printStackTrace();
            closeChannel(sc);
        }
    }

    private static class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object> {
        private final AsynchronousServerSocketChannel ssc;

        public AcceptHandler(AsynchronousServerSocketChannel ssc) {
            this.ssc = ssc;
        }

        @Override
        public void completed(AsynchronousSocketChannel sc, Object attachment) {
            try {
                System.out.printf("[%s] %s connected\n", Thread.currentThread().getName(), sc.getRemoteAddress());
            } catch (IOException e) {
                e.printStackTrace();
            }
            ByteBuffer buffer = ByteBuffer.allocate(16);
            // 讀事件由 ReadHandler 處理
            sc.read(buffer, buffer, new ReadHandler(sc));
            // 寫事件由 WriteHandler 處理
            sc.write(Charset.defaultCharset().encode("server hello!"), ByteBuffer.allocate(16), new WriteHandler(sc));
            // 處理完第一個 accpet 時授嘀，需要再次調(diào)用 accept 方法來處理下一個 accept 事件
            ssc.accept(null, this);
        }

        @Override
        public void failed(Throwable exc, Object attachment) {
            exc.printStackTrace();
        }
    }
}