• 首页 首页 icon
  • 工具库 工具库 icon
    • IP查询 IP查询 icon
  • 内容库 内容库 icon
    • 快讯库 快讯库 icon
    • 精品库 精品库 icon
    • 问答库 问答库 icon
  • 更多 更多 icon
    • 服务条款 服务条款 icon

Netty——NIO基础

武飞扬头像
啊噢1231
帮助1

目录

1、三大组件

1.1、Channel & Buffer

1.2、Selector

2、ByteBuffer

2.1、ByteBuffer的使用

2.2、ByteBuffer结构

2.3、ByteBuffer常用方法

2.4、字符串与ByteBuffer互转

2.5、Scattering Reads(分散读取)

2.6、GatheringWrites(集中写入)

2.7、综合案例——黏包、半包的处理

3、文件编程

3.1、FileChannel

3.2、两个Channel传输数据

3.3、Path

3.4、Files

4、网络编程

4.1、非阻塞 vs 阻塞

4.2、Selector-绑定Channel事件

4.3、Selector-监听Channel事件

4.4、Selector-何时不阻塞

4.5、处理消息的边界

4.5.1、ByteBuffer大小分配

4.6、处理可写事件

4.7、nio多线程优化

4.7.1、如何拿到CPU个数

5、NIO vs BIO

5.1、stream vs channel

5.2、IO模型

5.2.1、阻塞IO

5.2.2、非阻塞IO

5.2.3、多路复用

5.2.4、信号驱动

5.3、零拷贝

5.3.1、NIO优化

5.4、AIO


现在的互联网环境下,分布式系统大行其道,而分布式系统的根基在于网络编程,而Netty恰恰是Java领域网络编程的王者。如果要致力于开发高性能的服务器程序、高性能的客户端程序,必须掌握Netty,而本课程的目的就是带你进入基于Netty的网络编程世界。

1、三大组件

NIO(non-blocking io,非阻塞IO)。

1.1、Channel & Buffer

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

常见的Channel有:

  • FileChannel;
  • DatagramChannel;(UDP数据传输通道)
  • SocketChannel;(TCP数据传输通道)
  • ServerSocketChannel。

buffer则用来缓冲读写数据,常见的buffer有:

  • ByteBuffer:MappedByteBuffer,DirectByteBuffer,HeapByteBuffer;
  • ShortBuffer;
  • IntBuffer;
  • LongBuffer;
  • FloatBuffer;
  • DoubleBuffer;
  • CharBuffer。

1.2、Selector

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

学新通

学新通

学新通

2、ByteBuffer

2.1、ByteBuffer的使用

  1. 向buffer写入数据,例如调用channel.read(buffer);
  2. 调用flip()切换至读模式;
  3. 调用clear()或compact()切换至写模式;
  4. 重复1-4步骤。

2.2、ByteBuffer结构

ByteBuffer有以下重要属性:

  • capacity(容量);
  • position(位置);
  • limit(限制)。

写模式下,position是写入位置,limit等于容量capacity。flip动作发生后(切换为读模式),position切换为读取位置,limit切换为读取限制。clear动作发生后,状态position=0,limit=capacity。compact()方法,是把未读完的部分向前压缩,然后切换至写模式。

  1.  
    package com.clp.netty;
  2.  
     
  3.  
    import lombok.extern.slf4j.Slf4j;
  4.  
     
  5.  
    import java.io.FileInputStream;
  6.  
    import java.io.IOException;
  7.  
    import java.nio.ByteBuffer;
  8.  
    import java.nio.channels.FileChannel;
  9.  
     
  10.  
    @Slf4j
  11.  
    public class TestByteBuffer {
  12.  
    public static void main(String[] args) {
  13.  
    //File Channel
  14.  
    //1、输入输出流;2、RandomAccessFile
  15.  
    try (FileChannel channel = new FileInputStream("data.txt").getChannel()) {
  16.  
    //准备缓冲区
  17.  
    ByteBuffer buffer = ByteBuffer.allocate(3); //划分10个字节
  18.  
    while(true) {
  19.  
    //从channel读取数据,写入buffer
  20.  
    int len = channel.read(buffer); //返回值是读到的实际字节数
  21.  
    log.debug("读取到的字节数{}", len);
  22.  
    if(len == -1) break; //没有内容了
  23.  
    //打印buffer的内容
  24.  
    buffer.flip(); //切换到buffer的读模式
  25.  
    while(buffer.hasRemaining()) { //若还有剩余
  26.  
    byte b = buffer.get();
  27.  
    // System.out.println((char)b);
  28.  
    log.debug("实际字节{}", (char)b);
  29.  
    }
  30.  
    //切换到写模式
  31.  
    buffer.clear();
  32.  
    }
  33.  
    } catch (IOException e) {
  34.  
    }
  35.  
    }
  36.  
    }
学新通
  1.  
    package com.clp.netty;
  2.  
     
  3.  
    import java.nio.ByteBuffer;
  4.  
     
  5.  
    public class TestBufferReadWrite {
  6.  
    public static void main(String[] args) {
  7.  
    ByteBuffer buffer = ByteBuffer.allocate(10);
  8.  
    buffer.put((byte)0x61); //十六进制97
  9.  
    buffer.put(new byte[]{0x62, 0x63, 0x64}); //b,c,d
  10.  
     
  11.  
    System.out.println(buffer);
  12.  
    //System.out.println(buffer.get()); //读索引值为pos的位置的值(结果为0),同时pos还会往后移一位
  13.  
    System.out.println("------------");
  14.  
     
  15.  
    buffer.flip(); //切换读模式
  16.  
     
  17.  
    System.out.println(buffer);
  18.  
    System.out.println(buffer.get()); //读索引值为pos的位置的值(结果为97)
  19.  
    System.out.println("------------");
  20.  
     
  21.  
    buffer.compact();
  22.  
    System.out.println(buffer);
  23.  
    }
  24.  
    }
  25.  
     
  26.  
    结果:
  27.  
    java.nio.HeapByteBuffer[pos=4 lim=10 cap=10]
  28.  
    ------------
  29.  
    java.nio.HeapByteBuffer[pos=0 lim=4 cap=10]
  30.  
    97
  31.  
    ------------
  32.  
    java.nio.HeapByteBuffer[pos=3 lim=10 cap=10]
学新通

2.3、ByteBuffer常用方法

  1.  
    -分配空间,可以使用allocate()方法为ByteBuffer分配空间,其他buffer类也有该方法
  2.  
    ByteBuffer buf = ByteBuffer.allocate(16); //堆内存
  3.  
    ByteBuffer buf = ByteBuffer.allocateDirect(16); //直接内存
  4.  
    堆内存:读写效率较低,受到垃圾回收的影响。
  5.  
    直接内存:读写效率高(少一次数据的拷贝),不会受到垃圾回收的影响,分配的效率低。
  6.  
     
  7.  
    -向buffer写入数据
  8.  
    1、调用channel的read方法
  9.  
    int len = channel.read(buf); //返回值是读到的实际字节数
  10.  
    2、调用buffer自己的put方法
  11.  
    buf.put((byte)127);
  12.  
     
  13.  
    -从buffer读取数据到channel
  14.  
    1、调用channel的write()方法
  15.  
    int len = channel.write(buf);
  16.  
    2、调用buffer自己的get()方法
  17.  
    byte b = buf.get();
  18.  
    get()方法会让position读指针向后走,如果想重复读取数据:
  19.  
    可以调用rewind()方法将position重新置为0
  20.  
    或者调用get(int i)方法获取索引为i的内容,它不会移动指针。
  21.  
     
  22.  
    -mark() 和 reset()
  23.  
    mark是一个标记,记录position位置,reset是将pos重置到mark位置
学新通
  1.  
    package com.clp.netty;
  2.  
     
  3.  
    import lombok.extern.slf4j.Slf4j;
  4.  
    import java.nio.ByteBuffer;
  5.  
     
  6.  
    @Slf4j
  7.  
    public class TestByteBuffer {
  8.  
    public static void main(String[] args) {
  9.  
    ByteBuffer buf = ByteBuffer.allocate(10);
  10.  
    buf.put(new byte[]{'a', 'b', 'c', 'd'});
  11.  
    buf.flip(); //切换到读模式
  12.  
     
  13.  
    //从头开始读
  14.  
    buf.get(new byte[4]);
  15.  
    System.out.println(buf);
  16.  
    buf.rewind(); //将pos置为0
  17.  
    System.out.println((char)buf.get() " " buf);
  18.  
    buf.rewind();
  19.  
     
  20.  
    //mark & reset
  21.  
    //mark是一个标记,记录position位置,reset是将pos重置到mark位置
  22.  
    System.out.println((char)buf.get());
  23.  
    System.out.println((char)buf.get());
  24.  
    buf.mark(); //加标记,索引为2的位置
  25.  
    System.out.println((char)buf.get());
  26.  
    System.out.println((char)buf.get());
  27.  
    buf.reset(); //重置到索引为2的位置
  28.  
    System.out.println((char)buf.get());
  29.  
    System.out.println((char)buf.get());
  30.  
    buf.rewind();
  31.  
     
  32.  
    //get(i)方法。不会改变pos的值
  33.  
    System.out.println((char)buf.get(3) " " buf);
  34.  
    }
  35.  
    }
  36.  
     
  37.  
    结果:
  38.  
    java.nio.HeapByteBuffer[pos=4 lim=4 cap=10]
  39.  
    a java.nio.HeapByteBuffer[pos=1 lim=4 cap=10]
  40.  
    a
  41.  
    b
  42.  
    c
  43.  
    d
  44.  
    c
  45.  
    d
  46.  
    d java.nio.HeapByteBuffer[pos=0 lim=4 cap=10]
学新通

2.4、字符串与ByteBuffer互转

  1.  
    package com.clp.netty;
  2.  
     
  3.  
    import java.nio.ByteBuffer;
  4.  
    import java.nio.charset.Charset;
  5.  
    import java.nio.charset.StandardCharsets;
  6.  
     
  7.  
    public class TestByteBufferString {
  8.  
    public static void main(String[] args) {
  9.  
    //1、字符串转为ByteBuffer
  10.  
    ByteBuffer buffer = ByteBuffer.allocate(16);
  11.  
    buffer.put("hello".getBytes()); //默认操作系统的字符集,调用完之后仍为写模式
  12.  
    System.out.println(buffer);
  13.  
    String str = StandardCharsets.UTF_8.decode(buffer).toString();
  14.  
    System.out.println(str); //因为没有切换回读模式,因此打印失败
  15.  
    buffer.flip(); //手动切换为读模式
  16.  
    System.out.println(buffer);
  17.  
    System.out.println("---------------------------");
  18.  
     
  19.  
    //2、Charset
  20.  
    ByteBuffer buffer2 = StandardCharsets.UTF_8.encode("hello"); //自动切换回读模式
  21.  
    System.out.println(buffer2);
  22.  
    String str2 = StandardCharsets.UTF_8.decode(buffer2).toString();
  23.  
    System.out.println(str2); //因为buffer2已经切换回读模式,因此打印成功
  24.  
    System.out.println("---------------------------");
  25.  
     
  26.  
    //3、wrap
  27.  
    ByteBuffer buffer3 = ByteBuffer.wrap("hello".getBytes()); //自动切换回读模式
  28.  
    System.out.println(buffer3);
  29.  
    System.out.println("---------------------------");
  30.  
    }
  31.  
    }
  32.  
     
  33.  
    结果:
  34.  
    java.nio.HeapByteBuffer[pos=5 lim=16 cap=16]
  35.  
     
  36.  
    java.nio.HeapByteBuffer[pos=0 lim=16 cap=16]
  37.  
    ---------------------------
  38.  
    java.nio.HeapByteBuffer[pos=0 lim=5 cap=5]
  39.  
    hello
  40.  
    ---------------------------
  41.  
    java.nio.HeapByteBuffer[pos=0 lim=5 cap=5]
  42.  
    ---------------------------
学新通

2.5、Scattering Reads(分散读取)

  1.  
    package com.clp.netty;
  2.  
     
  3.  
    import java.io.FileNotFoundException;
  4.  
    import java.io.IOException;
  5.  
    import java.io.RandomAccessFile;
  6.  
    import java.nio.ByteBuffer;
  7.  
    import java.nio.channels.FileChannel;
  8.  
     
  9.  
    public class TestScatteringReads {
  10.  
    public static void main(String[] args) {
  11.  
    try {
  12.  
    FileChannel channel = new RandomAccessFile("3parts.txt", "r").getChannel();
  13.  
    ByteBuffer buf1 = ByteBuffer.allocate(3);
  14.  
    ByteBuffer buf2 = ByteBuffer.allocate(3);
  15.  
    ByteBuffer buf3 = ByteBuffer.allocate(5);
  16.  
    channel.read(new ByteBuffer[]{buf1, buf2, buf3});
  17.  
     
  18.  
    buf1.flip();buf2.flip();buf3.flip(); //切换为读模式
  19.  
    System.out.println(buf1);
  20.  
    System.out.println(buf2);
  21.  
    System.out.println(buf3);
  22.  
    } catch (FileNotFoundException e) {
  23.  
    e.printStackTrace();
  24.  
    } catch (IOException e) {
  25.  
    e.printStackTrace();
  26.  
    }
  27.  
    }
  28.  
    }
  29.  
     
  30.  
    结果:
  31.  
    java.nio.HeapByteBuffer[pos=0 lim=3 cap=3]
  32.  
    java.nio.HeapByteBuffer[pos=0 lim=3 cap=3]
  33.  
    java.nio.HeapByteBuffer[pos=0 lim=5 cap=5]
学新通

2.6、GatheringWrites(集中写入)

  1.  
    package com.clp.netty;
  2.  
     
  3.  
    import java.io.FileNotFoundException;
  4.  
    import java.io.IOException;
  5.  
    import java.io.RandomAccessFile;
  6.  
    import java.nio.ByteBuffer;
  7.  
    import java.nio.channels.FileChannel;
  8.  
    import java.nio.charset.StandardCharsets;
  9.  
     
  10.  
    public class TestGatheringWrites {
  11.  
    public static void main(String[] args) {
  12.  
    ByteBuffer buf1 = StandardCharsets.UTF_8.encode("hello"); //5B
  13.  
    ByteBuffer buf2 = StandardCharsets.UTF_8.encode("world"); //5B
  14.  
    ByteBuffer buf3 = StandardCharsets.UTF_8.encode("你好"); //6B
  15.  
     
  16.  
    //写入
  17.  
    try {
  18.  
    FileChannel channel = new RandomAccessFile("1part.txt", "rw").getChannel();
  19.  
    channel.write(new ByteBuffer[]{buf1, buf2, buf3});
  20.  
    } catch (FileNotFoundException e) {
  21.  
    e.printStackTrace();
  22.  
    } catch (IOException e) {
  23.  
    e.printStackTrace();
  24.  
    }
  25.  
    }
  26.  
    }
  27.  
     
  28.  
    结果:(1part.txt)
  29.  
    helloworld你好
学新通

2.7、综合案例——黏包、半包的处理

  1.  
    package com.clp.netty;
  2.  
     
  3.  
    import java.nio.ByteBuffer;
  4.  
     
  5.  
    public class TestByteBufferExam {
  6.  
    public static void main(String[] args) {
  7.  
    /**
  8.  
    * 网络上有许多条数据发送给服务器,数据之间用\n进行分隔,但由于某种原因这些数据在接收时,
  9.  
    * 被进行了重新组合,例如原始数据有3条为:
  10.  
    * HelloWorld\n
  11.  
    * I'm zhangsan\n
  12.  
    * How are you\n
  13.  
    * 变成了下面的两个bytebuffer(黏包、半包)
  14.  
    * Hello World\nI'm zhangsan\nHo
  15.  
    * w are you\n
  16.  
    * 现在要求你编写程序,将错乱的信息恢复成原始的按\n分隔的数据
  17.  
    */
  18.  
    //模拟网络编程
  19.  
    ByteBuffer source = ByteBuffer.allocate(32);
  20.  
    source.put("Hello,world\nI'm zhangsan\nHo".getBytes());
  21.  
    split(source);
  22.  
    source.put("w are you?\n".getBytes());
  23.  
    split(source);
  24.  
    }
  25.  
     
  26.  
    public static void split(ByteBuffer source) {
  27.  
    //切换成读模式
  28.  
    source.flip();
  29.  
     
  30.  
    for (int i = 0; i < source.limit(); i ) {
  31.  
    //找到完整消息
  32.  
    if(source.get(i) == '\n') {
  33.  
    int length = i 1 - source.position();
  34.  
    //把这条完成消息存入新的ByteBuffer
  35.  
    ByteBuffer target = ByteBuffer.allocate(length);
  36.  
    //从source读,向target写
  37.  
    for (int j = 0; j < length; j ) {
  38.  
    target.put(source.get());
  39.  
    }
  40.  
    target.flip();
  41.  
    System.out.println(target);
  42.  
    }
  43.  
    }
  44.  
     
  45.  
    //切换成写模式
  46.  
    source.compact(); //将剩余部分向前移动
  47.  
    }
  48.  
    }
  49.  
     
  50.  
    结果:
  51.  
    java.nio.HeapByteBuffer[pos=0 lim=12 cap=12]
  52.  
    java.nio.HeapByteBuffer[pos=0 lim=13 cap=13]
  53.  
    java.nio.HeapByteBuffer[pos=0 lim=13 cap=13]
学新通

3、文件编程

3.1、FileChannel

FileChannel只能工作在阻塞模式下。

  1.  
    -获取FileChannel
  2.  
    不能直接打开FileChannel,必须通过FileInputStream、FileOutputStream或者RandomAccessFile来获取FileChannel,他们都有getChannel()方法。
  3.  
    通过FileInputStream获取的channel只能读;
  4.  
    通过FileOutputChannel获取的channel只能写;
  5.  
    通过RandomAccessFile是否只能读写根据构造RandomAccessFile时的读写模式决定。
  6.  
     
  7.  
    -从FileChannel读取数据到ByteBuffer
  8.  
    会从channel读取数据填充ByteBuffer,返回值表示读到了多少字节,-1表示到达了文件的末尾。
  9.  
    int readBytes = channel.read(buffer);
  10.  
     
  11.  
    -正确写入数据到FileChannel方式:(SocketChannel)
  12.  
    ByteBuffer buffer = ...
  13.  
    buffer.put(..); //存入数据
  14.  
    buffer.flip(); //切换读模式
  15.  
     
  16.  
    while(buffer.hasRemaing()) { //检查buffer中是否有剩余数据
  17.  
    channel.write(buffer);
  18.  
    }
  19.  
    while中调用channel.write()是因为write()方法并不能保证一次将buffer中的内容全部写入channel。
  20.  
     
  21.  
    -关闭
  22.  
    channel必须关闭,不过调用了FileInputStream、FileOutputStream或者RandomAccessFile的close()方法回间接地调用channel地close()方法。
  23.  
     
  24.  
    -位置
  25.  
    获取当前位置:long pos = channel.position();
  26.  
    设置当前位置:channel.position(newPos);
  27.  
    设置当前位置时,如果设置为文件的末尾:
  28.  
    这时读取回返回-1
  29.  
    这时写入,回追加内容,但是要注意如果position超过了文件末尾,再写入时在新内容和原末尾之间会有空洞(00)。
  30.  
     
  31.  
    -大小
  32.  
    使用size()方法获取文件的大小。
  33.  
     
  34.  
    -强制写入
  35.  
    操作系统出于性能的考虑,会将数据缓存,不是立刻写入数据,可以调用force(true)方法将文件内容和元数据(文件的权限等信息)立刻写入磁盘。
学新通

3.2、两个Channel传输数据

  1.  
    package com.clp.netty;
  2.  
     
  3.  
    import java.io.*;
  4.  
    import java.nio.channels.FileChannel;
  5.  
     
  6.  
    public class Test {
  7.  
    public static void main(String[] args) {
  8.  
    try {
  9.  
    FileChannel from = new FileInputStream("from.txt").getChannel();
  10.  
    FileChannel to = new FileOutputStream("to.txt").getChannel();
  11.  
     
  12.  
    //效率高(底层利用操作系统的零拷贝进行优化),传输数据有上限(2G数据),可以多次传输来改进
  13.  
    long size = from.size();
  14.  
    for (long left = size; left>0;) {
  15.  
    long n = from.transferTo((size-left), left, to); //返回实际传输的字节数
  16.  
    left -= n;
  17.  
    }
  18.  
    } catch (FileNotFoundException e) {
  19.  
    e.printStackTrace();
  20.  
    } catch (IOException e) {
  21.  
    e.printStackTrace();
  22.  
    }
  23.  
    }
  24.  
    }
学新通

3.3、Path

jdk7引入了Path和Paths类。

  • Path用来表示文件路径;
  • Paths是工具类,用来获取Path实例。
  • .代表了当前路径;
  • ..代表了上一级路径。
  1.  
    Path source = Path.get("1.txt"); //相对路径,使用user.dir环境变量来定位1.txt
  2.  
    Path source = Path.get("d:\\1.txt"); //绝对路径,代表了d:\1.txt
  3.  
    Path source = Path.get("d:/1.txt"); //绝对路径,同样代表了d:\1.txt
  4.  
    Path source = Path.get("d:\\data", "projects"); //代表了d:\data\projects
  5.  
     
  6.  
    Path path = Path.get("d:\\data\\projects\\a\\..\\b");
  7.  
     
  8.  
    path.normalize(); //正常化路径

3.4、Files

  1.  
    -检查文件是否存在
  2.  
    Path path = Paths.get("helloworld/data.txt");
  3.  
    System.out.println(Files.exists(path));
  4.  
     
  5.  
    -创建一级目录
  6.  
    Path path = Path.get("helloworld");
  7.  
    Files.createDirectory(path);
  8.  
    如果目录已经存在,回抛异常FileAlreadyExistsException;
  9.  
    不能一次创建多级目录,否则会抛异常NoSuchFileException。
  10.  
     
  11.  
    -创建多级目录
  12.  
    Path path = Paths.get(""helloworld/d1/d2");
  13.  
    Files.createDirectories(path);
  14.  
     
  15.  
    -拷贝文件
  16.  
    Path source = Paths.get("helloworld/data.txt");
  17.  
    Path target = Paths.get("helloworld/target.txt");
  18.  
    Files.copy(source, target);
  19.  
    如果文件已经存在,会抛异常FileAlreadyExistsException;
  20.  
    如果希望用source覆盖target,需要用StandardCopyOption来控制:
  21.  
    Files.copy(source, target, StandardCopyOption.REPLACE_EXISTING);
  22.  
     
  23.  
    -移动文件
  24.  
    Path source = Paths.get("helloworld/data.txt");
  25.  
    Path target = Paths.get("helloworld/data.txt");
  26.  
    Files.move(source, target, StandardCopyOption.ATOMIC_MOVE);
  27.  
    StandardCopyOption.ATOMIC_MOVE保证文件移动的原子性。
  28.  
     
  29.  
    -删除文件
  30.  
    Path target = Paths.get("helloworld/target.txt");
  31.  
    Files.delete(target);
  32.  
    如果文件不存在,会抛异常NoSunchFileException。
  33.  
     
  34.  
    -删除目录
  35.  
    Path target = Paths.get("helloworld/d1");
  36.  
    Files.delete(target);
  37.  
    如果目录还有内容,会抛异常DirectoryNotEmptyException。
  38.  
     
  39.  
    -遍历目录和文件
  40.  
    package com.clp.netty;
  41.  
     
  42.  
    import java.io.*;
  43.  
    import java.nio.file.*;
  44.  
    import java.nio.file.attribute.BasicFileAttributes;
  45.  
    import java.util.concurrent.atomic.AtomicInteger;
  46.  
     
  47.  
    public class Test {
  48.  
    public static void main(String[] args) throws IOException {
  49.  
    //不能使用局部变量来使用,故使用AtomicInteger
  50.  
    AtomicInteger dirCount = new AtomicInteger();
  51.  
    AtomicInteger fileCount = new AtomicInteger();
  52.  
    //从c:\\my开始遍历
  53.  
    Files.walkFileTree(Paths.get("C:\\project"), new SimpleFileVisitor<Path>() {
  54.  
    @Override
  55.  
    public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
  56.  
    System.out.println("=====>" dir);
  57.  
    dirCount.incrementAndGet();
  58.  
     
  59.  
    return super.preVisitDirectory(dir, attrs);
  60.  
    }
  61.  
     
  62.  
    @Override
  63.  
    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
  64.  
    //检查文件是否是以.jar结尾
  65.  
    if(file.toString().endsWith(".jar")) {
  66.  
    System.out.println(file);
  67.  
    }
  68.  
    fileCount.incrementAndGet();
  69.  
     
  70.  
    return super.visitFile(file, attrs);
  71.  
    }
  72.  
    });
  73.  
    System.out.println("dir count:" dirCount);
  74.  
    System.out.println("file count:" fileCount);
  75.  
    }
  76.  
    }
  77.  
     
  78.  
    -删除多级目录
  79.  
    package com.clp.netty;
  80.  
     
  81.  
    import java.io.*;
  82.  
    import java.nio.file.*;
  83.  
    import java.nio.file.attribute.BasicFileAttributes;
  84.  
    import java.util.concurrent.atomic.AtomicInteger;
  85.  
     
  86.  
    public class Test {
  87.  
    public static void main(String[] args) throws IOException {
  88.  
    Files.walkFileTree(Paths.get("C:\\Drivers"), new SimpleFileVisitor<Path>() {
  89.  
    @Override //进入目录之前
  90.  
    public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
  91.  
    System.out.println("====>进入" dir);
  92.  
     
  93.  
    return super.preVisitDirectory(dir, attrs);
  94.  
    }
  95.  
     
  96.  
    @Override //访问文件
  97.  
    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
  98.  
    System.out.println("访问文件:" file);
  99.  
     
  100.  
    return super.visitFile(file, attrs);
  101.  
    }
  102.  
     
  103.  
    @Override //退出目录之后
  104.  
    public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
  105.  
    System.out.println("====>退出" dir);
  106.  
     
  107.  
    return super.postVisitDirectory(dir, exc);
  108.  
    }
  109.  
    });
  110.  
    }
  111.  
    }
  112.  
     
  113.  
    -文件夹的复制
  114.  
    package com.clp.netty;
  115.  
     
  116.  
    import java.io.*;
  117.  
    import java.nio.file.*;
  118.  
     
  119.  
    public class Test {
  120.  
    public static void main(String[] args) throws IOException {
  121.  
    String source = "C:\\source";
  122.  
    String target = "C:\\target";
  123.  
    Files.walk(Paths.get(source)).forEach(path -> {
  124.  
    //将target目录名替换为source目录名,返回最终的名字
  125.  
    String targetName = path.toString().replace(source, target);
  126.  
    if(Files.isDirectory(path)) {
  127.  
    //是目录,就创建目录
  128.  
    try {
  129.  
    Files.createDirectory(Paths.get(targetName));
  130.  
    } catch (IOException e) {
  131.  
    e.printStackTrace();
  132.  
    }
  133.  
    }else if(Files.isRegularFile(path)) {
  134.  
    //是文件,就复制文件
  135.  
    try {
  136.  
    Files.copy(path, Paths.get(targetName));
  137.  
    } catch (IOException e) {
  138.  
    e.printStackTrace();
  139.  
    }
  140.  
    }
  141.  
    });
  142.  
    }
  143.  
    }
学新通

4、网络编程

4.1、非阻塞 vs 阻塞

阻塞:

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

非阻塞:

  1.  
    非阻塞模式下,相关方法都不会让线程暂停。
  2.  
    在ServerChannel.accept()在没有连接建立时,会返回null,继续运行;
  3.  
    SocketChannel.read()在没有数据可读时,会返回0,但线程不必阻塞,可以去执行其它SocketChannel的read()或是去执行ServerSocketChannel.accept();
  4.  
    写数据时,线程只是等待数据写入Channel即可,无需等Channel通过网络把数据发送出去。
  5.  
    但非阻塞模式下,即使没有连接建立,和可读数据,线程仍然在不断运行,白白浪费了cpu;
  6.  
    数据复制过程中,线程实际还是阻塞的(AIO改进的地方)。

多路复用:

  1.  
    单线程可以配合Selector完成对多个Channel可读写事件的监控,这称之为多路复用。
  2.  
    多路复用仅针对网络IO,普通文件IO没法利用多路复用。
  3.  
    如果不用Selector的非阻塞模式,线程大部分时间都在做无用功,而Selector能够保证:
  4.  
    有可连接事件时才去连接;
  5.  
    有可读事件才去读取;
  6.  
    有可写事件才去写入。
  7.  
    限于网络传输能力,Channel未必时时可写,一旦Channel可写,会触发Selector的可写事件。

非阻塞模式:

  1.  
    package com.clp.netty2;
  2.  
     
  3.  
    import lombok.extern.slf4j.Slf4j;
  4.  
     
  5.  
    import java.io.IOException;
  6.  
    import java.net.InetSocketAddress;
  7.  
    import java.nio.ByteBuffer;
  8.  
    import java.nio.channels.*;
  9.  
    import java.nio.charset.Charset;
  10.  
    import java.util.ArrayList;
  11.  
    import java.util.Iterator;
  12.  
    import java.util.List;
  13.  
     
  14.  
    @Slf4j
  15.  
    public class Server {
  16.  
    public static void main(String[] args) throws IOException {
  17.  
    //*创建selector,可以管理多个channel
  18.  
    Selector selector = Selector.open();
  19.  
     
  20.  
    //使用nio来理解阻塞模式,单线程
  21.  
    //0、创建ByteBuffer
  22.  
    ByteBuffer buffer = ByteBuffer.allocate(16);
  23.  
    //1、创建服务器
  24.  
    ServerSocketChannel ssc = ServerSocketChannel.open();
  25.  
    ssc.configureBlocking(false); //切换成非阻塞模式,会影响accept()方法。
  26.  
     
  27.  
    //*建立selector和channel的联系(注册)
  28.  
    //*SelectionKey就是将来事件发生后,通过它可以知道事件和哪个channel的事件
  29.  
    /**
  30.  
    * 事件:
  31.  
    * accept:会在有连接请求时触发
  32.  
    * connect:在客户端连接建立后触发
  33.  
    * read:可读事件
  34.  
    * write:可写事件
  35.  
    */
  36.  
    SelectionKey sscKey = ssc.register(selector, 0, null);
  37.  
    sscKey.interestOps(SelectionKey.OP_ACCEPT); //指明sscKey只关注accept事件
  38.  
     
  39.  
    //2、绑定监听端口
  40.  
    ssc.bind(new InetSocketAddress(8080));
  41.  
     
  42.  
    /**************************使用selector**************************/
  43.  
    while(true) {
  44.  
    //3、select()方法,没有事件发生,线程阻塞;有事件,线程才会恢复运行
  45.  
    //select在事件未处理时,它不会阻塞。事件发生后要么处理,要么取消,不能置之不理。
  46.  
    selector.select();
  47.  
     
  48.  
    //4、处理事件,selectedKeys内部包含了所有发生的事件
  49.  
    //selector会在发生事件后,向selectedKeys集合中加入这个key,但不会删除。
  50.  
    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
  51.  
    //遍历事件集合
  52.  
    while(iter.hasNext()) {
  53.  
    SelectionKey key = iter.next();
  54.  
    //处理key时,要从selectedKeys集合中删除
  55.  
    iter.remove(); //移除这个key,不能用增强for,因为不能在遍历时删除
  56.  
     
  57.  
    log.debug("key:{}", key);
  58.  
    //5、区分事件类型
  59.  
    if(key.isAcceptable()) { //如果是accept事件
  60.  
    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
  61.  
    SocketChannel sc = channel.accept();//建立连接
  62.  
    // key.cancel(); //若不想处理事件,则可以取消事件,则下次select.select()时不会阻塞
  63.  
    sc.configureBlocking(false);
  64.  
    //将sc注册到selector上,这样selector可以管理这个channel的事件
  65.  
    SelectionKey scKey = sc.register(selector, 0, null);
  66.  
    scKey.interestOps(SelectionKey.OP_READ); //关注读事件
  67.  
    }else if(key.isReadable()) { //如果是read事件
  68.  
    try {
  69.  
    SocketChannel channel = (SocketChannel) key.channel();//拿到触发事件的channel
  70.  
    ByteBuffer buffer1 = ByteBuffer.allocate(4);
  71.  
    //客户端异常/正常断开,都会产生一个读事件。
  72.  
    int read = channel.read(buffer1); //如果是正常断开,返回值是-1
  73.  
    if(read == -1) key.cancel(); //若正常断开,则将这个key从selector中取消(删除)
  74.  
    else {
  75.  
    buffer1.flip(); //切换为读模式
  76.  
    System.out.println(Charset.defaultCharset().decode(buffer1));
  77.  
    }
  78.  
    } catch (IOException e) {
  79.  
    e.printStackTrace();
  80.  
    //客户端异常/正常断开,都会产生一个读事件。
  81.  
    key.cancel(); //将这个key从selector中取消(删除),因为客户端关闭了
  82.  
    }
  83.  
    }
  84.  
    }
  85.  
    }
  86.  
     
  87.  
    /**************************不使用selector**************************/
  88.  
    // //3、accept,建立与客户端的连接。
  89.  
    // List<SocketChannel> channels = new ArrayList<SocketChannel>();
  90.  
    // while(true) {
  91.  
    // //log.debug("connecting...");
  92.  
    // //accept()若为阻塞方法,意味着线程在此等待,直到连接建立以后继续执行;
  93.  
    // //若为非阻塞,线程还会继续执行。如果没有连接建立,则sc为null。
  94.  
    // SocketChannel sc = ssc.accept(); //4、SocketChannel数据读写的通道,用来与客户端进行通信
  95.  
    // if(sc != null) {
  96.  
    // log.debug("connected..");
  97.  
    // sc.configureBlocking(false); //非阻塞模式,会影响read()方法
  98.  
    // channels.add(sc);
  99.  
    // }
  100.  
    //
  101.  
    // //5、接收客户端发送的数据
  102.  
    // for(SocketChannel channel : channels) {
  103.  
    // //log.debug("before read...{}", channel);
  104.  
    // //若为阻塞方法,线程停止运行,知道客户端发送新数据才会继续执行
  105.  
    // //若为非阻塞方法,线程仍然会继续运行;如果没有读到数据,read返回0。
  106.  
    // if(channel.read(buffer)>0) {
  107.  
    // buffer.flip(); //切换到读模式
  108.  
    // System.out.println(buffer);
  109.  
    // buffer.clear(); //切换到写模式
  110.  
    // log.debug("after read...{}",channel);
  111.  
    // }
  112.  
    // }
  113.  
    // }
  114.  
    }
  115.  
    }
学新通
  1.  
    package com.clp.netty2;
  2.  
     
  3.  
    import java.io.IOException;
  4.  
    import java.net.InetSocketAddress;
  5.  
    import java.nio.channels.SocketChannel;
  6.  
    import java.nio.charset.Charset;
  7.  
     
  8.  
    public class Client {
  9.  
    public static void main(String[] args) throws IOException {
  10.  
    SocketChannel sc = SocketChannel.open();
  11.  
    sc.connect(new InetSocketAddress("localhost", 8080));
  12.  
    System.out.println("waiting...");
  13.  
    sc.write(Charset.defaultCharset().encode("hello,你好"));
  14.  
    }
  15.  
    }
学新通

4.2、Selector-绑定Channel事件

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

4.3、Selector-监听Channel事件

  1.  
    可以通过下面三种方法来监听是否有事件发生,方法的返回值代表有多少channel发生了事件。
  2.  
     
  3.  
    方法1:阻塞直到绑定事件发生。
  4.  
    int count = selector.select();
  5.  
    方法2:阻塞直到绑定事件发生,或是超时(时间单位为ms)。
  6.  
    int count = selector.select(long timeout);
  7.  
    方法3:不会阻塞,也就是不管有没有事件,立刻返回,自己根据返回值检查是否有事件。
  8.  
    int count = selector.selectNow();

4.4、Selector-何时不阻塞

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

4.5、处理消息的边界

  • 一种思维是固定消息长度,数据包大小一样,服务器按预定长度读取,缺点是浪费带宽。
  • 另一种思路是按分隔符(如"\n")拆分,缺点是效率低;
  • TLV格式,即Type类型、Length长度、Value数据,类型和长度已知的情况下,就可以方便获取信息大小,分配合适的buffer,缺点是buffer需要提前分配,如果内容过大,则影响server吞吐量。(Http 1.1是TLV格式,Http 2.0是LTV格式)。

学新通

  1.  
    package com.clp.test;
  2.  
     
  3.  
    import lombok.extern.slf4j.Slf4j;
  4.  
     
  5.  
    import java.io.IOException;
  6.  
    import java.net.InetSocketAddress;
  7.  
    import java.nio.ByteBuffer;
  8.  
    import java.nio.channels.SelectionKey;
  9.  
    import java.nio.channels.Selector;
  10.  
    import java.nio.channels.ServerSocketChannel;
  11.  
    import java.nio.channels.SocketChannel;
  12.  
    import java.nio.charset.Charset;
  13.  
    import java.util.Iterator;
  14.  
     
  15.  
    @Slf4j
  16.  
    public class Server {
  17.  
    private static void split(ByteBuffer source) {
  18.  
    source.flip(); //开启读模式
  19.  
    for (int i = 0; i < source.limit(); i ) {
  20.  
    //找到一条完整消息
  21.  
    if(source.get(i) == '\n') {
  22.  
    int length = i 1 - source.position();
  23.  
    //把这条完整消息存入新的ByteBuffer
  24.  
    ByteBuffer target = ByteBuffer.allocate(length);
  25.  
    //从source读,向target写
  26.  
    for (int j = 0; j < length; j ) {
  27.  
    target.put(source.get());
  28.  
    }
  29.  
    System.out.println(target);
  30.  
    }
  31.  
    }
  32.  
    source.compact(); //压缩
  33.  
    }
  34.  
     
  35.  
    public static void main(String[] args) throws IOException {
  36.  
    //*创建selector,可以管理多个channel
  37.  
    Selector selector = Selector.open();
  38.  
     
  39.  
    //使用nio来理解阻塞模式,单线程
  40.  
    //0、创建ByteBuffer
  41.  
    ByteBuffer buffer = ByteBuffer.allocate(16);
  42.  
    //1、创建服务器
  43.  
    ServerSocketChannel ssc = ServerSocketChannel.open();
  44.  
    ssc.configureBlocking(false); //切换成非阻塞模式,会影响accept()方法。
  45.  
     
  46.  
    //*建立selector和channel的联系(注册)
  47.  
    //*SelectionKey就是将来事件发生后,通过它可以知道事件和哪个channel的事件
  48.  
    /**
  49.  
    * 事件:
  50.  
    * accept:会在有连接请求时触发
  51.  
    * connect:在客户端连接建立后触发
  52.  
    * read:可读事件
  53.  
    * write:可写事件
  54.  
    */
  55.  
    SelectionKey sscKey = ssc.register(selector, 0, null);
  56.  
    sscKey.interestOps(SelectionKey.OP_ACCEPT); //指明sscKey只关注accept事件
  57.  
     
  58.  
    //2、绑定监听端口
  59.  
    ssc.bind(new InetSocketAddress(8080));
  60.  
     
  61.  
    /**************************使用selector**************************/
  62.  
    while(true) {
  63.  
    //3、select()方法,没有事件发生,线程阻塞;有事件,线程才会恢复运行
  64.  
    //select在事件未处理时,它不会阻塞。事件发生后要么处理,要么取消,不能置之不理。
  65.  
    selector.select();
  66.  
     
  67.  
    //4、处理事件,selectedKeys内部包含了所有发生的事件
  68.  
    //selector会在发生事件后,向selectedKeys集合中加入这个key,但不会删除。
  69.  
    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
  70.  
    //遍历事件集合
  71.  
    while(iter.hasNext()) {
  72.  
    SelectionKey key = iter.next();
  73.  
    //处理key时,要从selectedKeys集合中删除
  74.  
    iter.remove(); //移除这个key,不能用增强for,因为不能在遍历时删除
  75.  
     
  76.  
    log.debug("key:{}", key);
  77.  
    //5、区分事件类型
  78.  
    if(key.isAcceptable()) { //如果是accept事件
  79.  
    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
  80.  
    SocketChannel sc = channel.accept();//建立连接
  81.  
    // key.cancel(); //若不想处理事件,则可以取消事件,则下次select.select()时不会阻塞
  82.  
    sc.configureBlocking(false);
  83.  
    ByteBuffer buffer1 = ByteBuffer.allocate(4); //attachment:附件
  84.  
    //将sc注册到selector上,这样selector可以管理这个channel的事件
  85.  
    //将一个ByteBuffer作为附件关联到selectionKey上
  86.  
    SelectionKey scKey = sc.register(selector, 0, buffer1);
  87.  
    scKey.interestOps(SelectionKey.OP_READ); //关注读事件
  88.  
    }else if(key.isReadable()) { //如果是read事件
  89.  
    try {
  90.  
    SocketChannel channel = (SocketChannel) key.channel();//拿到触发事件的channel
  91.  
    ByteBuffer buf = (ByteBuffer) key.attachment();//
  92.  
    //客户端异常/正常断开,都会产生一个读事件。
  93.  
    int read = channel.read(buf); //如果是正常断开,返回值是-1
  94.  
    if(read == -1) key.cancel(); //若正常断开,则将这个key从selector中取消(删除)
  95.  
    else {
  96.  
    split(buf);
  97.  
    if(buf.position() == buf.limit()) {
  98.  
    //说明buf没有被压缩,即空间不够存一个消息,需要扩容
  99.  
    ByteBuffer newBuffer = ByteBuffer.allocate(buf.capacity()*2);
  100.  
    buf.flip(); //切换成读模式
  101.  
    newBuffer.put(buf); //将旧内容写入新buffer
  102.  
    key.attach(newBuffer); //附件替换,newBuffer替换原来的buf,达到扩容的目的
  103.  
    }
  104.  
    }
  105.  
     
  106.  
    } catch (IOException e) {
  107.  
    e.printStackTrace();
  108.  
    //客户端异常/正常断开,都会产生一个读事件。
  109.  
    key.cancel(); //将这个key从selector中取消(删除),因为客户端关闭了
  110.  
    }
  111.  
    }
  112.  
    }
  113.  
    }
  114.  
    }
  115.  
    }
学新通
  1.  
    package com.clp.test;
  2.  
     
  3.  
    import java.io.IOException;
  4.  
    import java.net.InetSocketAddress;
  5.  
    import java.nio.channels.SocketChannel;
  6.  
    import java.nio.charset.Charset;
  7.  
     
  8.  
    public class Client {
  9.  
    public static void main(String[] args) throws IOException {
  10.  
    SocketChannel sc = SocketChannel.open();
  11.  
    sc.connect(new InetSocketAddress("localhost", 8080));
  12.  
    sc.write(Charset.defaultCharset().encode("0123\n456789abcdef\n"));
  13.  
    System.in.read();
  14.  
    }
  15.  
    }
学新通

4.5.1、ByteBuffer大小分配

  • 每一个channel都需要记录可能被切分的消息,因为ByteBuffer不能被多个channel共同使用,不是线程安全的,因此需要为每个channel维护一个独立的ByteBuffer;
  • ByteBuffer不能太大,比如一个ByteBuffer 1Mb的话,要支持百万连接就要1 Tb内存,因此需要设计大小可变的ByteBuffer。

如何设计大小可变的ByteBuffer?

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

4.6、处理可写事件

  1.  
    package com.clp.test;
  2.  
     
  3.  
    import java.io.IOException;
  4.  
    import java.net.InetSocketAddress;
  5.  
    import java.nio.ByteBuffer;
  6.  
    import java.nio.channels.SelectionKey;
  7.  
    import java.nio.channels.Selector;
  8.  
    import java.nio.channels.ServerSocketChannel;
  9.  
    import java.nio.channels.SocketChannel;
  10.  
    import java.nio.charset.Charset;
  11.  
    import java.util.Iterator;
  12.  
     
  13.  
    public class WriteServer {
  14.  
    public static void main(String[] args) throws IOException {
  15.  
    ServerSocketChannel ssc = ServerSocketChannel.open();
  16.  
    ssc.configureBlocking(false); //设置为非阻塞模式
  17.  
     
  18.  
    Selector selector = Selector.open();
  19.  
    ssc.register(selector, SelectionKey.OP_ACCEPT); //注册并关注accept事件
  20.  
     
  21.  
    ssc.bind(new InetSocketAddress(8080)); //监听端口
  22.  
     
  23.  
    while(true) {
  24.  
    selector.select();
  25.  
    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
  26.  
    if(iter.hasNext()) {
  27.  
    SelectionKey key = iter.next();
  28.  
    iter.remove();
  29.  
    if(key.isAcceptable()) {
  30.  
    SocketChannel sc = ssc.accept();
  31.  
    sc.configureBlocking(false);
  32.  
    SelectionKey scKey = sc.register(selector, 0, null);
  33.  
    scKey.interestOps(SelectionKey.OP_READ); //关注可读事件
  34.  
     
  35.  
    //1、向客户端发送大量数据
  36.  
    StringBuilder sb = new StringBuilder();
  37.  
    for (int i = 0; i < 3000000; i ) {
  38.  
    sb.append("a");
  39.  
    }
  40.  
    ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());
  41.  
     
  42.  
    //2、返回值代表实际写入的字节数
  43.  
    int write = sc.write(buffer);
  44.  
    System.out.println(write);
  45.  
    //判断buffer是否还有剩余内容
  46.  
    if(buffer.hasRemaining()) {
  47.  
    //既关注读事件(1),又关注写事件(4),若为5,说明既读又写事件
  48.  
    scKey.interestOps(scKey.interestOps() SelectionKey.OP_WRITE);
  49.  
    //scKey.interestOps(scKey.interestOps() | SelectionKey.OP_WRITE);
  50.  
    //把未写完的数据挂到scKey上
  51.  
    scKey.attach(buffer);
  52.  
    }
  53.  
    }else if(key.isWritable()) {
  54.  
    ByteBuffer buf = (ByteBuffer) key.attachment();
  55.  
    SocketChannel sc = (SocketChannel) key.channel();
  56.  
    //返回值代表实际写入的字节数
  57.  
    int write = sc.write(buf);
  58.  
    System.out.println(write);
  59.  
    //清理操作
  60.  
    if(!buf.hasRemaining()) {
  61.  
    //buf为空,说明写完了
  62.  
    key.attach(null); //清除buffer附件
  63.  
    key.interestOps(key.interestOps() - SelectionKey.OP_WRITE); //不需要再关注可写事件
  64.  
    }
  65.  
    }
  66.  
    }
  67.  
    }
  68.  
    }
  69.  
    }
学新通
  1.  
    package com.clp.test;
  2.  
     
  3.  
    import java.io.IOException;
  4.  
    import java.net.InetSocketAddress;
  5.  
    import java.nio.ByteBuffer;
  6.  
    import java.nio.channels.SocketChannel;
  7.  
     
  8.  
    public class WriteClient {
  9.  
    public static void main(String[] args) throws IOException {
  10.  
    SocketChannel sc = SocketChannel.open();
  11.  
    sc.connect(new InetSocketAddress("localhost", 8080));
  12.  
     
  13.  
    //接收数据
  14.  
    int count = 0;
  15.  
    while(true) {
  16.  
    ByteBuffer buffer = ByteBuffer.allocate(1024);
  17.  
    count = sc.read(buffer);
  18.  
    System.out.println(count);
  19.  
    buffer.clear(); //清空buffer
  20.  
    }
  21.  
    }
  22.  
    }
学新通

4.7、nio多线程优化

利用多线程优化:现在都是多核CPU,设计时要充分考虑别让CPU的力量被白白浪费。

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

分两组选择器:

        单线程配一个选择器,专门处理accept事件;

        创建CPU核心数的线程,每个线程配一个选择器,轮流处理read事件。

4.7.1、如何拿到CPU个数

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

这个问题直到jdk 10才修复,使用jvm参数UserContainerSupport配置,默认开启。

5、NIO vs BIO

5.1、stream vs channel

  • stream不会自动缓冲数据,channel会利用系统提供的发送缓冲区、接收缓冲区(更为底层);
  • stream仅支持阻塞API,channel同时支持阻塞、非阻塞API、网络channel可配合selector实现多路复用;
  • 二者均为全双工,即读写可以同时进行。

5.2、IO模型

同步阻塞、同步非阻塞、同步多路复用、异步阻塞(不存在这种情况)、异步非阻塞。

  • 同步:线程自己去获取结果一个(线程);
  • 异步:线程自己不去获取结果,而是由其他线程送结果(至少两个线程)。

当调用一次channel.rea()或stream.read()后,会切换至操作系统内核来完成真正数据读取,而读取又分为两个阶段,分别为:

  • 等待数据阶段;
  • 复制数据阶段。

5.2.1、阻塞IO

5.2.2、非阻塞IO

5.2.3、多路复用

5.2.4、信号驱动

5.3、零拷贝

传统IO问题:

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

  1.  
    File f = new File("helloworld/data.txt");
  2.  
    RandomAccessFile file = new RandomAccessFile(file, 'r');
  3.  
     
  4.  
    byte[] buf = new byte[(int)f.length()];
  5.  
    file.read(buf);
  6.  
     
  7.  
    Socket socket = ..;
  8.  
    socket.getOutputStream().write(buf);

学新通

  • Java本身不具备IO读写能力,因此read()方法调用后,要从Java程序的用户态切换至核心态,去调用操作系统(Kernel)的读能力,将数据读入内核缓冲区,这期间用户线程阻塞,操作系统使用DMA(Direct Memory Access)来实现文件读,期间也不会使用CPU。
  • 从内核态切换回用户态,将数据从内核缓冲区读入用户缓冲区(即byte[] buf),这期间CPU会参与拷贝,无法利用DMA。
  • 调用write()方法,这时将数据从用户缓冲区(byte[] buf)写入socket,CPU会参与拷贝;
  • 接下来要向网卡写数据,这项能力Java又不具备,因此又得从用户态切换至核心态,嗲用操作系统的写能力,使用DMA将socket缓冲区的数据写入网卡,不会使用CPU。

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

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

5.3.1、NIO优化

1、通过DirectByteBuf

  • ByteBuffer.allocate(10); //HeapBuffer,使用的还是java。
  • ByteBuffer.allocateDirect(10); //DirectByteBuffer,使用的是操作系统内存。

学新通

大部分步骤与优化前相同,不再赘述,唯有一点:Java可以使用DirectByteBuf将堆外内存映射到JVM内存中来直接访问使用。

  • 这块内存不受jvm垃圾回收的影响,因此内存地址固定,有助于IO读写;
  • Java中的DirectByteBuf对象仅维护了此内存的虚引用,内存回收分成两步:① DirectByteBuf对象被垃圾回收,将虚引用加入引用队列;② 通过专门线程访问引用队列,根据虚引用释放堆外内存。
  • 减少了一次数据拷贝,用户态与内核态的切换次数没有减少。

2、进一步优化(底层采用了linux 2.1后提供的sendFile()方法),Java中对应着两个channel调用transferTo() / transferFrom()方法拷贝数据。

学新通

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

可以看到

  • 只发生了一次用户态与内核态的切换;
  • 数据拷贝了3次。

3、进一步优化(linux 2.4)

学新通

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

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

  • 更少的用户态与内核态的转换;
  • 不利用CPU计算,减少CPU缓存伪共享;
  • 零拷贝适合小文件传输。

5.4、AIO

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

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

异步模型需要底层操作系统(Kernel)提供支持:

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

这篇好文章是转载于:学新通技术网

  • 版权申明: 本站部分内容来自互联网,仅供学习及演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,请提供相关证据及您的身份证明,我们将在收到邮件后48小时内删除。
  • 本站站名: 学新通技术网
  • 本文地址: /boutique/detail/tanhfhhhjk
系列文章
更多 icon
同类精品
更多 icon
继续加载