Netty基础入门——文件编程、网络编程【2】

秒速五厘米 2024-03-30 15:07 125阅读 0赞

Netty基础入门——文件编程、网络编程【2】

基础入门【1】

1 文件编程

1.1 channel

  1. 两个channel传输数据

transferTo方法一次性最多传输2G大小的文件，如果超出会丢弃

  1. public static void main(String[] args) {
  3.     try (
  4.         FileChannel from = new FileInputStream("from.txt").getChannel();
  5.         FileChannel to = new FileOutputStream("to.txt").getChannel();
  6.         ) {
  8.         //传输文件【从0位置开始，传输from.size()大小的数据，传输到to文件】
  9.         from.transferTo(0, from.size(), to);
  10.     } catch (IOException e) {
  12.         e.printStackTrace();
  13.     }
  14. }

超过2G大小的文件传输

  1. public static void main(String[] args) {
  3.    try (
  4.        FileChannel from = new FileInputStream("from.txt").getChannel();
  5.        FileChannel to = new FileOutputStream("to.txt").getChannel();
  6.        ) {
  8.        //此种方式传输效率高，系统底层会利用操作系统的零拷贝进行优化
  9.        long size = from.size();
  10.        //left 代表还剩多少字节
  11.        for(long left = size; left > 0;){
  13.            System.out.println("position:" + (size - left) + ",left:" + left);
  14.            //起始位置：size - left
  15.            left -= from.transferTo((size - left), left, to);
  16.        }
  17.    } catch (IOException e) {
  19.        e.printStackTrace();
  20.    }
  21. }

1.2 Path、Paths、Files

jdk7 引入了 Path 和 Paths 类

  • Path 用来表示文件路径
  • Paths 是工具类，用来获取 Path 实例
①检查文件是否存在
  1. // Path path = Paths.get("test/from.txt");
  2. Path path = Paths.get("test\\from.txt");
  3. System.out.println(Files.exists(path));
②创建多级目录
  1. Path path = Paths.get("data/test");
  2. System.out.println(Files.createDirectories(path));
③拷贝文件
  1. Path from = Paths.get("test/from.txt");
  2. Path to = Paths.get("data/to.txt");
  3. Files.copy(from, to);
  4. //如果文件已存在，会抛异常 FileAlreadyExistsException
  5. //如果希望用 source 覆盖掉 target，需要用 StandardCopyOption 来控制
  6. //Files.copy(source, target, StandardCopyOption.REPLACE_EXISTING);
④移动文件
  1. Path from = Paths.get("test/from.txt");
  2.  Path to = Paths.get("data/to.txt");
  3.  Files.move(from, to, StandardCopyOption.ATOMIC_MOVE);
⑤删除文件、目录
  1. //删除文件
  2. Path from = Paths.get("test/from.txt");
  3. Files.delete(from);
  5. //删除目录
  6. Path dir = Paths.get("test");
  7. Files.delete(dir);
⑥遍历文件夹、统计特定文件个数

观察者模式

  • 遍历文件夹

    private static void walkDir() throws IOException {

    1. Path path = Paths.get("D:\\系统默认\\桌面\\Yi-music\\music-server\\src\\main\\java\\com\\zi\\music");
    2. AtomicInteger dirCount = new AtomicInteger();
    3. AtomicInteger fileCount = new AtomicInteger();
    4. //遍历文件夹
    5. Files.walkFileTree(path, new SimpleFileVisitor<Path>(){
    7.     //遍历目录
    8.     @Override
    9.     public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
    11.         System.out.println(dir);
    12.         dirCount.incrementAndGet();
    13.         return super.preVisitDirectory(dir, attrs);
    14.     }
    16.     //遍历文件
    17.     @Override
    18.     public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
    20.         System.out.println(file);
    21.         fileCount.incrementAndGet();
    22.         return super.visitFile(file, attrs);
    23.     }
    24. });
    25. System.out.println(dirCount);
    26. System.out.println(fileCount);

    }

  • 统计.java文件个数

    private static void countAbsFile() throws IOException {

    1. Path path = Paths.get("D:\\系统默认\\桌面\\Yi-music\\music-server\\src\\main\\java\\com\\zi\\music");
    2. AtomicInteger javaCount = new AtomicInteger();
    3. //遍历文件夹
    4. Files.walkFileTree(path, new SimpleFileVisitor<Path>(){
    6.     @Override
    7.     public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
    9.         if(file.toFile().getName().endsWith(".java")){
    11.             javaCount.incrementAndGet();
    12.         }
    13.         return super.visitFile(file, attrs);
    14.     }
    15. });
    16. System.out.println("javaCount: " + javaCount);

    }

⑦删除目录、拷贝目录
  • 删除目录

注意：删除的目录一定要是没有重要数据的文件夹，通过以下代码删除的方式，不走回收站，直接系统删除

  1. private static void deleteDir() throws IOException {
  3.     Path path = Paths.get("d:\\a");
  4.     Files.walkFileTree(path, new SimpleFileVisitor<Path>(){
  6.         @Override
  7.         public FileVisitResult visitFile(Path file, BasicFileAttributes attrs)
  8.                 throws IOException {
  10.             Files.delete(file);
  11.             return super.visitFile(file, attrs);
  12.         }
  14.         @Override
  15.         public FileVisitResult postVisitDirectory(Path dir, IOException exc)
  16.                 throws IOException {
  18.             Files.delete(dir);
  19.             return super.postVisitDirectory(dir, exc);
  20.         }
  21.     });
  22. }

  • 拷贝目录

    private static void copyDir() throws IOException {

    1. long start = System.currentTimeMillis();
    2. String source = "D:\\Snipaste-1.16.2-x64";
    3. String target = "D:\\Snipaste-1.16.2-x64aaa";
    5. Files.walk(Paths.get(source)).forEach(path -> {
    7.     try {
    9.         String targetName = path.toString().replace(source, target);
    10.         // 是目录
    11.         if (Files.isDirectory(path)) {
    13.             Files.createDirectory(Paths.get(targetName));
    14.         }
    15.         // 是普通文件
    16.         else if (Files.isRegularFile(path)) {
    18.             Files.copy(path, Paths.get(targetName));
    19.         }
    20.     } catch (IOException e) {
    22.         e.printStackTrace();
    23.     }
    24. });
    25. long end = System.currentTimeMillis();
    26. System.out.println(end - start);

    }

2 网络编程

2.1 阻塞模式

阻塞模式下，下面方法都会造成线程暂停

  • ServerSocketChannel.accept 会在没有连接建立时让线程暂停
  • SocketChannel.read 会在没有数据可读时让线程暂停
    阻塞：线程暂停，线程不占用cpu，但是相当于线程闲置
  • 单线程下，阻塞方法之间相互影响，几乎不能正常工作，需要多线程支持

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

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

demo测试：
server:

  1. @Slf4j
  2. public class Server {
  5.     public static void main(String[] args) throws IOException {
  7.         //0.分配缓冲区u
  8.         ByteBuffer buffer = ByteBuffer.allocate(16);
  9.         //1. 创建服务器
  10.         ServerSocketChannel ssc = ServerSocketChannel.open();
  11.         //2. 绑定监听端口
  12.         ssc.bind(new InetSocketAddress(8888));
  13.         //3. 连接集合
  14.         List<SocketChannel> channels = new ArrayList<>();
  15.         while(true){
  17.             //4. accept建立与客户端连接， SocketChannel用来与客户端通信
  18.             log.debug("connecting...");
  19.             SocketChannel sc = ssc.accept();//阻塞方法，线程停止运行
  20.             log.debug("connected...{}", sc);
  21.             channels.add(sc);
  22.             for(SocketChannel channel : channels){
  24.                 //5. 接收客户端发送的数据
  25.                 log.debug("before read...{}", channel);
  26.                 channel.read(buffer);//阻塞方法，线程停止运行
  27.                 buffer.flip();//切换模式
  28.                 buffer.clear();
  29.                 log.debug("after read...{}", channel);
  30.             }
  31.         }
  32.     }
  33. }

Client:

  1. @Slf4j
  2. public class Client {
  4.     public static void main(String[] args) throws IOException {
  6.         SocketChannel sc = SocketChannel.open();
  7.         sc.connect(new InetSocketAddress("localhost", 8888));
  8.         System.out.println("waiting....");
  9.     }
  10. }
  1. 启动两个客户端和服务器端
    在这里插入图片描述

虽然启动了两个客户端，但是服务器只接收到一个，因为是阻塞的，需要等到第一个客户端连接处理完了才会轮到下一个

  1. client写入数据，让服务器处理第一个连接

选中sc变量，alt+F8，evaluate

写入数据：

  1. sc.write(StandardCharsets.UTF_8.encode("hello"))

在这里插入图片描述

  1. 结果：
    服务器已经由54250切换到处理54254
    在这里插入图片描述

如果你的idea无法同时打开两个run运行台

  • 新版：
    在这里插入图片描述
  • 旧版：
    在这里插入图片描述

2.2 非阻塞模式

非阻塞模式下，accept和read都不会让线程暂停

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

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

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

阻塞 -> 非阻塞：

  1. //1. 创建服务器
  2. ServerSocketChannel ssc = ServerSocketChannel.open();
  3. //调整为非阻塞模式
  4. ssc.configureBlocking(false);
  5. ....
  6. SocketChannel sc = ssc.accept();//阻塞方法，线程停止运行
  7. //改为非阻塞
  8. sc.configureBlocking(false);

Server：

  1. @Slf4j
  2. public class Server {
  5.     public static void main(String[] args) throws IOException {
  7.         //0.分配缓冲区u
  8.         ByteBuffer buffer = ByteBuffer.allocate(16);
  9.         //1. 创建服务器
  10.         ServerSocketChannel ssc = ServerSocketChannel.open();
  11.         ssc.configureBlocking(false);        //非阻塞
  12.         //2. 绑定监听端口
  13.         ssc.bind(new InetSocketAddress(8888));
  14.         //3. 连接集合
  15.         List<SocketChannel> channels = new ArrayList<>();
  16.         while(true){
  18.             //4. accept建立与客户端连接， SocketChannel用来与客户端通信
  19.             SocketChannel sc = ssc.accept();//阻塞方法，线程停止运行
  20.             if(sc != null){
  22.                 //如果没有客户端连接，sc为null
  23.                 log.debug("connected...{}", sc);
  24.                 sc.configureBlocking(false);    //非阻塞
  25.                 channels.add(sc);
  26.             }
  27.             for(SocketChannel channel : channels){
  29.                 //5. 接收客户端发送的数据
  30.                 int read = channel.read(buffer);//非阻塞，线程仍然会继续运行，如果没有读到数据，read返回0
  31.                 if(read > 0){
  33.                     //切换为读模式
  34.                     buffer.flip();
  35.                     buffer.clear();
  36.                     log.debug("after read...{}", channel);
  37.                 }
  38.             }
  39.         }
  40.     }
  41. }

Client端代码不变：

  1. @Slf4j
  2. public class Client {
  4.     public static void main(String[] args) throws IOException {
  6.         SocketChannel sc = SocketChannel.open();
  7.         sc.connect(new InetSocketAddress("localhost", 8888));
  8.         System.out.println("waiting....");
  9.     }
  10. }

测试：

与上面操作一样，启动一个Server和两个Client，可以发现，虽然第一个Client没有发送数据，但是Server依然可以处理两个Client

在这里插入图片描述

2.3 多路复用与Selector

2.3.1 多路复用

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

  • 多路复用仅针对网络IO、普通文件IO没法利用多路复用
  • Selector可以保证：
  1. 有可连接事件时才去连接
  2. 有可读事件才去读取
  3. 有可写事件才去写入【受限于网络传输能力，只有当Channel可写时才会处方Selector的可写事件】
  1. selector四种可绑定的事件类型：
  2. 1. accept - 会在有连接请求时触发
  3. 2. connect - 是客户端，连接建立后触发
  4. 3. read - 可读事件
  5. 4. write - 可写事件
2.3.2 Selector

selector 版

selector

thread

channel

channel

channel

①好处：

  • 一个线程配合 selector 就可以监控多个 channel 的事件，事件发生线程才去处理。避免非阻塞模式下所做无用功
  • 让这个线程能够被充分利用
  • 节约了线程的数量
  • 减少了线程上下文切换

②使用步骤

  1. 1. 创建Selector
  2. Selector selector = Selector.open();
  3. 2. 绑定Channel事件（注册事件）
  4. channel.configureBlocking(false);
  5. SelectionKey key = channel.register(selector, 绑定事件);
  6. 3. 监听Channel事件（方法的返回值代表有多少 channel 发生了事件）
  7. int count = selector.select();
  8. //int count = selector.select(long timeout); //阻塞直到绑定事件发生，或是超时单位：ms
  9. //int count = selector.selectNow();//不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件
  10. 4. 处理事件

③selector什么时候不会阻塞

  • 事件发生时

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

④处理事件

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

cancel 会取消注册在 selector 上的 channel，并从 keys 集合中删除 key 后续不会再监听事件

iter.remove()；处理完事件后需要移除，否则会报NPE

[1] 处理accept事件

Client:

  1. @Slf4j
  2. public class Client {
  4.     public static void main(String[] args) throws IOException {
  6.         try (Socket socket = new Socket("localhost", 8888)) {
  8.             System.out.println(socket);
  9.             socket.getOutputStream().write("hello".getBytes(StandardCharsets.UTF_8));
  10.             System.in.read();
  11.         }catch (Exception e){
  13.             e.printStackTrace();
  14.         }
  15.     }
  16. }

Server:

  1. @Slf4j
  2. public class Server {
  5.     public static void main(String[] args) throws IOException {
  7.         try (ServerSocketChannel channel = ServerSocketChannel.open()) {
  9.             channel.bind(new InetSocketAddress(8888));
  10.             System.out.println(channel);
  11.             Selector selector = Selector.open();
  12.             //非阻塞
  13.             channel.configureBlocking(false);
  14.             //处理连接事件[注册事件]
  15.             //           accept
  16.             // selector ---------> channel
  17.             channel.register(selector, SelectionKey.OP_ACCEPT);
  18.             while(true){
  20.                 int count = selector.select();
  21.                 log.debug("select count:{}", count);
  22.                 //获取所有事件
  23.                 Set<SelectionKey> keys = selector.selectedKeys();
  24.                 //遍历所有事件，逐一处理
  25.                 Iterator<SelectionKey> iter = keys.iterator();
  26.                 while(iter.hasNext()){
  28.                     SelectionKey key = iter.next();
  29.                     //判断事件类型
  30.                     if(key.isAcceptable()){
  32.                         ServerSocketChannel c = (ServerSocketChannel) key.channel();
  33.                         //必须处理：事件只能被处理或者撤销
  34.                         SocketChannel sc = c.accept();
  35.                         log.debug("{}", sc);
  36.                     }
  37.                     //处理完毕之后，必须将事件移除，否则会NPE
  38.                     iter.remove();
  39.                 }
  40.             }
  41.         }catch (IOException e){
  43.             e.printStackTrace();
  44.         }
  45.     }
  46. }
[2]处理read事件
  • Client端代码同accept事件的客户端

  • Server端：

    @Slf4j
    public class Server {

  1.     public static void main(String[] args) throws IOException {
  3.         try (ServerSocketChannel channel = ServerSocketChannel.open()) {
  5.             channel.bind(new InetSocketAddress(8888));
  6.             System.out.println(channel);
  7.             Selector selector = Selector.open();
  8.             //非阻塞
  9.             channel.configureBlocking(false);
  10.             //处理连接事件[注册事件]
  11.             //           accept
  12.             // selector ---------> channel
  13.             channel.register(selector, SelectionKey.OP_ACCEPT);
  14.             while(true){
  16.                 int count = selector.select();
  17.                 log.debug("select count:{}", count);
  18.                 //获取所有事件
  19.                 Set<SelectionKey> keys = selector.selectedKeys();
  20.                 //遍历所有事件，逐一处理
  21.                 Iterator<SelectionKey> iter = keys.iterator();
  22.                 while(iter.hasNext()){
  24.                     SelectionKey key = iter.next();
  25.                     //判断事件类型
  26.                     if(key.isAcceptable()){
  28.                         ServerSocketChannel c = (ServerSocketChannel) key.channel();
  29.                         //必须处理：事件只能被处理或者撤销
  30.                         SocketChannel sc = c.accept();
  31.                         sc.configureBlocking(false);
  32.                         //           read
  33.                         // selector ---------> channel
  34.                         sc.register(selector, SelectionKey.OP_READ);
  35.                         log.debug("连接已建立：{}", sc);
  36.                     }else if(key.isReadable()){
  38.                         SocketChannel sc = (SocketChannel) key.channel();
  39.                         ByteBuffer buffer = ByteBuffer.allocate(128);
  40.                         int read = sc.read(buffer);
  41.                         if(read == -1){
  43.                             key.cancel();
  44.                             sc.close();
  45.                         }else {
  47.                             buffer.flip();
  48.                         }
  49.                     }
  50.                     //处理完毕之后，必须将事件移除
  51.                     iter.remove();
  52.                 }
  53.             }
  54.         }catch (IOException e){
  56.             e.printStackTrace();
  57.         }
  58.     }
  59. }

在这里插入图片描述

[3]消息边界处理

法一：固定消息长度

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

法二：按指定分隔符拆分（如：换行符）

按分隔符拆分，缺点是效率低

法三：TLV格式（type、length、value）

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

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

以方式二为例：

netty也是类似的处理方式，不过netty底层的bytebuffer是自适应的（可以扩容、缩容）

客户端1服务器ByteBuffer1ByteBuffer2发送 01234567890abcdef3333\r第一次 read 存入 01234567890abcdef扩容拷贝 01234567890abcdef第二次 read 存入 3333\r01234567890abcdef3333\r客户端1服务器ByteBuffer1ByteBuffer2

Server：

  1. @Slf4j
  2. public class Server {
  5.     public static void main(String[] args) throws IOException {
  7.         //1. 创建selector， 管理多个channel
  8.         Selector selector = Selector.open();
  9.         ServerSocketChannel ssc = ServerSocketChannel.open();
  10.         ssc.configureBlocking(false);//非阻塞
  11.         //2. 建立selector和channel的联系（注册）
  12.         //SelectionKey就是将来事件发生后，通过它可以知道事件来自哪个channel
  13.         SelectionKey sscKey = ssc.register(selector, 0, null);
  14.         //设置key只关注accept事件
  15.         sscKey.interestOps(SelectionKey.OP_ACCEPT);
  16.         log.debug("sscKey:{}", sscKey);
  17.         ssc.bind(new InetSocketAddress(8888));//channel监听端口
  18.         while(true){
  20.             //3. select方法，没有事件发生，线程阻塞，有事件，线程才会恢复运行
  21.             //select 再事件未处理时，它不会阻塞事件发生后要么处理，要么取消，不能置之不理
  22.             selector.select();
  23.             //4. 处理事件，selectedKeys内部包含了所有发生的事件
  24.             Iterator<SelectionKey> iter = selector.selectedKeys().iterator();//accept、read
  25.             while (iter.hasNext()) {
  27.                 SelectionKey key = iter.next();
  28.                 //处理key时，要从selectedKeys集合中删除，否则下次处理会有问题【客户端正常、异常断开】
  29.                 iter.remove();
  30.                 log.debug("key:{}", key);
  31.                 //5. 区分事件类型
  32.                 if(key.isAcceptable()){
  34.                     ServerSocketChannel channel = (ServerSocketChannel) key.channel();
  35.                     SocketChannel sc = channel.accept();
  36.                     sc.configureBlocking(false);
  37.                     ByteBuffer buffer = ByteBuffer.allocate(16);//attachement
  38.                     //将一个byteBuffer作为附件关联到selectionKey上
  39.                     SelectionKey scKey = sc.register(selector, 0, buffer);
  40.                     scKey.interestOps(SelectionKey.OP_READ);//关注read事件
  41.                     log.debug("{}", sc);
  42.                     log.debug("scKey:{}", scKey);
  43.                 }else if(key.isReadable()){
  45.                     try {
  47.                         SocketChannel channel = (SocketChannel) key.channel();//拿到触发事件的channel
  48.                         //获取selectionKey的关联附件
  49.                         ByteBuffer buffer = (ByteBuffer) key.attachment();
  50.                         int read = channel.read(buffer);//如果是客户端正常断开， read方法的返回值是-1
  51.                         if(read == -1){
  53.                             key.cancel();
  54.                         }else {
  56.                             //对数据进行处理，防止粘包半包
  57.                             split(buffer);
  58.                             //需要扩容[一个buffer存放不下一条完整的数据]
  59.                             if(buffer.position() == buffer.limit()){
  61.                                 ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
  62.                                 buffer.flip();
  63.                                 newBuffer.put(buffer);// 0123456789abcdef3333\n
  64.                                 key.attach(newBuffer);
  65.                             }
  66.                         }
  67.                     } catch (IOException e){
  69.                         e.printStackTrace();
  70.                         key.cancel();//因为客户端断开了，因此需要将key取消（从selector的keys集合中真正删除key）
  71.                     }
  72.                 }
  73.             }
  74.         }
  75.     }
  77.     /**
  78.      * 处理粘包半包问题
  79.      * @param source
  80.      */
  81.     private static void split(ByteBuffer source){
  83.         //limit = position  position = 0【切换为读模式】
  84.         source.flip();
  85.         for(int i = 0; i < source.limit(); i++){
  87.             //根据规定字符找到一条完整消息【'\n'】
  88.             if(source.get(i) == '\n'){
  90.                 //fad998877fa\n221  [source.position()当前读取到得位置]
  91.                 int length = i + 1 - source.position();
  92.                 //将这条完整消息存入新的ByteBuffer
  93.                 ByteBuffer target = ByteBuffer.allocate(length);
  94.                 //从source读，向target写
  95.                 for(int j = 0; j < length; j++){
  97.                     target.put(source.get());
  98.                 }
  99.                 debugAll(target);
  100.             }
  101.         }
  102.         source.compact(); // 0123456789abcdef  position 16 limit 16
  103.     }
  105. }

Client

  1. @Slf4j
  2. public class Client {
  4.     public static void main(String[] args) throws IOException {
  6.         SocketChannel sc = SocketChannel.open();
  7.         sc.connect(new InetSocketAddress("localhost", 8888));
  8.         SocketAddress address = sc.getLocalAddress();
  9.         // sc.write(Charset.defaultCharset().encode("hello\nworld\n"));
  10.         sc.write(Charset.defaultCharset().encode("0123\n456789abcdef"));
  11.         sc.write(Charset.defaultCharset().encode("0123456789abcdef3333\n"));
  12.         System.in.read();
  13.     }
  14. }
[4]ByteBuffer大小分配
  • 每个channel都需要维护一个独立的ByteBuffer

  • ByteBuffer不能太大

    • 方法1：首先分配一个较小的buffer，例如4K，如果数据不够再分配8k的buffer，将4k的内容拷贝到8k的buffer上【消息连续易处理，但是数据拷贝耗费性能】
    • 方法2：用多个数组组成buffer，一个数组不够，就把多出来的写入新数组，【避免了拷贝引起的性能损耗，但是消息存储不连续解析复杂】
2.3.3 处理write事件（一次写不完）
  • 非阻塞模式下，无法保证把 buffer 中所有数据都写入 channel，因此需要追踪 write 方法的返回值（代表实际写入字节数）

  • 用 selector 监听所有 channel 的可写事件，每个 channel 都需要一个 key 来跟踪 buffer，但这样又会导致占用内存过多，就有两阶段策略

    • 当消息处理器第一次写入消息时，才将 channel 注册到 selector 上
    • selector 检查 channel 上的可写事件，如果所有的数据写完了，就取消 channel 的注册
    • 如果不取消，会每次可写均会触发 write 事件

服务端：

  1. public class WriteServer {
  4.     public static void main(String[] args) throws IOException {
  6.         ServerSocketChannel ssc = ServerSocketChannel.open();
  7.         ssc.configureBlocking(false);//设置非阻塞
  8.         ssc.bind(new InetSocketAddress(8888));
  10.         Selector selector = Selector.open();
  11.         //注册selector到ssc，同时监听accept事件
  12.         ssc.register(selector, SelectionKey.OP_ACCEPT);
  13.         while(true){
  15.             selector.select();
  16.             //拿到绑定到selector上的所有key
  17.             Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
  18.             while(iter.hasNext()){
  20.                 SelectionKey key = iter.next();
  21.                 iter.remove();
  22.                 if(key.isAcceptable()){
  24.                     //ssc只有一个，因此可以直接获取
  25.                     SocketChannel sc = ssc.accept();
  26.                     sc.configureBlocking(false);
  27.                     SelectionKey scKey = sc.register(selector, SelectionKey.OP_READ);
  28.                     //1. 向客户端发送内容
  29.                     StringBuilder sb = new StringBuilder();
  30.                     for (int i = 0; i < 30000000; i++) {
  32.                         sb.append("a");
  33.                     }
  34.                     ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());
  35.                     int write = sc.write(buffer);
  36.                     //3. write：实际写了多少字节
  37.                     System.out.println("实际写入字节：" + write);
  38.                     //4. 如果有剩余未读字节，才需要关注写事件
  39.                     if(buffer.hasRemaining()){
  41.                         //read：1        write：4
  42.                         //在原有关注事件的基础上 + 额外关注写事件
  43.                         scKey.interestOps(scKey.interestOps() + SelectionKey.OP_WRITE);
  44.                         //把buffer作为附件加入sckey
  45.                         scKey.attach(buffer);
  46.                     }
  47.                 }else if(key.isWritable()){
  49.                     //拿到附件
  50.                     ByteBuffer buffer = (ByteBuffer) key.attachment();
  51.                     SocketChannel sc = (SocketChannel) key.channel();
  52.                     int write = sc.write(buffer);
  53.                     System.out.println("实际写入字节数：" + write);
  54.                     if(!buffer.hasRemaining()){
  55.     //没有剩余字节，写完了
  56.                         key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);
  57.                         key.attach(null);
  58.                     }
  59.                 }
  60.             }
  61.         }
  62.     }
  64. }

客户端：

  1. public class WriteClient {
  3.     public static void main(String[] args) throws IOException {
  5.         Selector selector = Selector.open();
  6.         SocketChannel sc = SocketChannel.open();
  7.         sc.configureBlocking(false);
  8.         sc.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
  9.         sc.connect(new InetSocketAddress("localhost", 8888));
  10.         int count = 0; //统计接收到得数据量
  11.         while(true){
  13.             selector.select();
  14.             Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
  15.             while(iter.hasNext()){
  17.                 SelectionKey key = iter.next();
  18.                 iter.remove();//拿到key之后要remove，否则会重复消费
  19.                 if(key.isConnectable()){
  21.                     System.out.println(sc.finishConnect());
  22.                 }else if(key.isReadable()){
  24.                     ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
  25.                     count += sc.read(buffer);
  26.                     buffer.clear();
  27.                     System.out.println("总共接收到得数据量：" + count);
  28.                 }
  29.             }
  30.         }
  31.     }
  32. }

实现多次读写：
在这里插入图片描述

发表评论

表情：
评论列表 （有 0 条评论，125人围观）

还没有评论，来说两句吧...

相关阅读