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

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

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

[基础入门【1】][1]

### 1 文件编程 ###

#### 1.1 channel ####

1.  两个channel传输数据

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

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

超过2G大小的文件传输

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

#### 1.2 Path、Paths、Files ####

> jdk7 引入了 Path 和 Paths 类
> 
>  *  Path 用来表示文件路径
>  *  Paths 是工具类，用来获取 Path 实例

##### ①检查文件是否存在 #####

// Path path = Paths.get("test/from.txt");
    Path path = Paths.get("test\\from.txt");
    System.out.println(Files.exists(path));

##### ②创建多级目录 #####

Path path = Paths.get("data/test");
    System.out.println(Files.createDirectories(path));

##### ③拷贝文件 #####

Path from = Paths.get("test/from.txt");
    Path to = Paths.get("data/to.txt");
    Files.copy(from, to);
    //如果文件已存在，会抛异常 FileAlreadyExistsException
    //如果希望用 source 覆盖掉 target，需要用 StandardCopyOption 来控制
    //Files.copy(source, target, StandardCopyOption.REPLACE_EXISTING);

##### ④移动文件 #####

Path from = Paths.get("test/from.txt");
     Path to = Paths.get("data/to.txt");
     Files.move(from, to, StandardCopyOption.ATOMIC_MOVE);

##### ⑤删除文件、目录 #####

//删除文件
    Path from = Paths.get("test/from.txt");
    Files.delete(from);
    
    //删除目录
    Path dir = Paths.get("test");
    Files.delete(dir);

##### ⑥遍历文件夹、统计特定文件个数 #####

> 观察者模式

*  遍历文件夹

private static void walkDir() throws IOException {
        
        Path path = Paths.get("D:\\系统默认\\桌面\\Yi-music\\music-server\\src\\main\\java\\com\\zi\\music");
        AtomicInteger dirCount = new AtomicInteger();
        AtomicInteger fileCount = new AtomicInteger();
        //遍历文件夹
        Files.walkFileTree(path, new SimpleFileVisitor<Path>(){
        
    		//遍历目录
            @Override
            public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
        
                System.out.println(dir);
                dirCount.incrementAndGet();
                return super.preVisitDirectory(dir, attrs);
            }
    
    		//遍历文件
            @Override
            public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
        
                System.out.println(file);
                fileCount.incrementAndGet();
                return super.visitFile(file, attrs);
            }
        });
        System.out.println(dirCount);
        System.out.println(fileCount);
    }

*  统计.java文件个数

private static void countAbsFile() throws IOException {
        
        Path path = Paths.get("D:\\系统默认\\桌面\\Yi-music\\music-server\\src\\main\\java\\com\\zi\\music");
        AtomicInteger javaCount = new AtomicInteger();
        //遍历文件夹
        Files.walkFileTree(path, new SimpleFileVisitor<Path>(){
        
            @Override
            public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
        
                if(file.toFile().getName().endsWith(".java")){
        
                    javaCount.incrementAndGet();
                }
                return super.visitFile(file, attrs);
            }
        });
        System.out.println("javaCount: " + javaCount);
    }

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

*  删除目录

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

private static void deleteDir() throws IOException {
        
        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);
            }
        });
    }

*  拷贝目录

private static void copyDir() throws IOException {
        
        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);
    }

### 2 网络编程 ###

#### 2.1 阻塞模式 ####

> 阻塞模式下，下面方法都会造成线程暂停
> 
>  *  ServerSocketChannel.accept 会在没有连接建立时让线程暂停
>  *  SocketChannel.read 会在没有数据可读时让线程暂停  
>     阻塞：线程暂停，线程不占用cpu，但是相当于线程闲置

*  单线程下，阻塞方法之间相互影响，几乎不能正常工作，需要多线程支持
 *  但多线程下，有新的问题，体现在以下方面
    
     *  32 位 jvm 一个线程 320k，64 位 jvm 一个线程 1024k，如果连接数过多，必然导致 OOM，并且线程太多，反而会因为频繁上下文切换导致性能降低
     *  可以采用线程池技术来减少线程数和线程上下文切换，但治标不治本，如果有很多连接建立，但长时间 inactive，会阻塞线程池中所有线程，因此不适合长连接，只适合短连接

demo测试：  
`server:`

@Slf4j
    public class Server {
        
    
        public static void main(String[] args) throws IOException {
        
            //0.分配缓冲区u
            ByteBuffer buffer = ByteBuffer.allocate(16);
            //1. 创建服务器
            ServerSocketChannel ssc = ServerSocketChannel.open();
            //2. 绑定监听端口
            ssc.bind(new InetSocketAddress(8888));
            //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. 接收客户端发送的数据
                    log.debug("before read...{}", channel);
                    channel.read(buffer);//阻塞方法，线程停止运行
                    buffer.flip();//切换模式
                    buffer.clear();
                    log.debug("after read...{}", channel);
                }
            }
        }
    }

`Client:`

1.  启动两个客户端和服务器端  
    ![在这里插入图片描述][4cd985ec60694743b35c24cccefde768.png]

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

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

> 选中sc变量，alt+F8，evaluate

写入数据：

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

![在这里插入图片描述][cd64f89aef634fd38575bd84a31721ab.png]  
3. 结果：  
服务器已经由54250切换到处理54254  
![在这里插入图片描述][d796ccf5a1a740dc8b9cd1f49b6f0504.png]

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

*  新版：  
    ![在这里插入图片描述][a671f9ddd7594639a938051083c98e63.png]
 *  旧版：  
    ![在这里插入图片描述][c5ed6e12242d4af5bfd5f0610caa3729.png]

#### 2.2 非阻塞模式 ####

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

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

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

阻塞 -> 非阻塞：

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

Server：

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

Client端代码不变：

测试：

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

![在这里插入图片描述][2ebb1a801e4b419ba62227e46a197a76.png]

#### 2.3 多路复用与Selector ####

##### 2.3.1 多路复用 #####

> 单线程可以配合Selector完成对多个Channel可读写事件的监控，这称之为多路复用
> 
>  *  多路复用仅针对网络IO、普通文件IO没法利用多路复用
>  *  Selector可以保证：
> 
> 1.  有可连接事件时才去连接
> 2.  有可读事件才去读取
> 3.  有可写事件才去写入【受限于网络传输能力，只有当Channel可写时才会处方Selector的可写事件】

selector四种可绑定的事件类型：
    1. accept - 会在有连接请求时触发
    2. connect - 是客户端，连接建立后触发
    3. read - 可读事件
    4. write - 可写事件

##### 2.3.2 Selector #####

selector 版

selector

thread

channel

①好处：

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

②使用步骤

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

@Slf4j
    public class Client {
        
        public static void main(String[] args) throws IOException {
        
            try (Socket socket = new Socket("localhost", 8888)) {
        
                System.out.println(socket);
                socket.getOutputStream().write("hello".getBytes(StandardCharsets.UTF_8));
                System.in.read();
            }catch (Exception e){
        
                e.printStackTrace();
            }
        }
    }

Server:

@Slf4j
    public class Server {
        
    
        public static void main(String[] args) throws IOException {
        
            try (ServerSocketChannel channel = ServerSocketChannel.open()) {
        
                channel.bind(new InetSocketAddress(8888));
                System.out.println(channel);
                Selector selector = Selector.open();
                //非阻塞
                channel.configureBlocking(false);
                //处理连接事件[注册事件]
                //           accept
                // selector ---------> channel
                channel.register(selector, SelectionKey.OP_ACCEPT);
                while(true){
        
                    int count = selector.select();
                    log.debug("select count:{}", count);
                    //获取所有事件
                    Set<SelectionKey> keys = selector.selectedKeys();
                    //遍历所有事件，逐一处理
                    Iterator<SelectionKey> iter = keys.iterator();
                    while(iter.hasNext()){
        
                        SelectionKey key = iter.next();
                        //判断事件类型
                        if(key.isAcceptable()){
        
                            ServerSocketChannel c = (ServerSocketChannel) key.channel();
                            //必须处理：事件只能被处理或者撤销
                            SocketChannel sc = c.accept();
                            log.debug("{}", sc);
                        }
                        //处理完毕之后，必须将事件移除，否则会NPE
                        iter.remove();
                    }
                }
            }catch (IOException e){
        
                e.printStackTrace();
            }
        }
    }

###### \[2\]处理read事件 ######

*  Client端代码同accept事件的客户端
 *  Server端：

@Slf4j
    public class Server {
        
    
        public static void main(String[] args) throws IOException {
        
            try (ServerSocketChannel channel = ServerSocketChannel.open()) {
        
                channel.bind(new InetSocketAddress(8888));
                System.out.println(channel);
                Selector selector = Selector.open();
                //非阻塞
                channel.configureBlocking(false);
                //处理连接事件[注册事件]
                //           accept
                // selector ---------> channel
                channel.register(selector, SelectionKey.OP_ACCEPT);
                while(true){
        
                    int count = selector.select();
                    log.debug("select count:{}", count);
                    //获取所有事件
                    Set<SelectionKey> keys = selector.selectedKeys();
                    //遍历所有事件，逐一处理
                    Iterator<SelectionKey> iter = keys.iterator();
                    while(iter.hasNext()){
        
                        SelectionKey key = iter.next();
                        //判断事件类型
                        if(key.isAcceptable()){
        
                            ServerSocketChannel c = (ServerSocketChannel) key.channel();
                            //必须处理：事件只能被处理或者撤销
                            SocketChannel sc = c.accept();
                            sc.configureBlocking(false);
                            //           read
                            // selector ---------> channel
                            sc.register(selector, SelectionKey.OP_READ);
                            log.debug("连接已建立：{}", sc);
                        }else if(key.isReadable()){
        
                            SocketChannel sc = (SocketChannel) key.channel();
                            ByteBuffer buffer = ByteBuffer.allocate(128);
                            int read = sc.read(buffer);
                            if(read == -1){
        
                                key.cancel();
                                sc.close();
                            }else {
        
                                buffer.flip();
                            }
                        }
                        //处理完毕之后，必须将事件移除
                        iter.remove();
                    }
                }
            }catch (IOException e){
        
                e.printStackTrace();
            }
        }
    }

![在这里插入图片描述][0d11da36ce414224ac2a41c52f61bf30.png]

###### \[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：

@Slf4j
    public class Server {
        
    
        public static void main(String[] args) throws IOException {
        
            //1. 创建selector， 管理多个channel
            Selector selector = Selector.open();
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.configureBlocking(false);//非阻塞
            //2. 建立selector和channel的联系（注册）
            //SelectionKey就是将来事件发生后，通过它可以知道事件来自哪个channel
            SelectionKey sscKey = ssc.register(selector, 0, null);
            //设置key只关注accept事件
            sscKey.interestOps(SelectionKey.OP_ACCEPT);
            log.debug("sscKey:{}", sscKey);
            ssc.bind(new InetSocketAddress(8888));//channel监听端口
            while(true){
        
                //3. select方法，没有事件发生，线程阻塞，有事件，线程才会恢复运行
                //select 再事件未处理时，它不会阻塞事件发生后要么处理，要么取消，不能置之不理
                selector.select();
                //4. 处理事件，selectedKeys内部包含了所有发生的事件
                Iterator<SelectionKey> iter = selector.selectedKeys().iterator();//accept、read
                while (iter.hasNext()) {
        
                    SelectionKey key = iter.next();
                    //处理key时，要从selectedKeys集合中删除，否则下次处理会有问题【客户端正常、异常断开】
                    iter.remove();
                    log.debug("key:{}", key);
                    //5. 区分事件类型
                    if(key.isAcceptable()){
        
                        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                        SocketChannel sc = channel.accept();
                        sc.configureBlocking(false);
                        ByteBuffer buffer = ByteBuffer.allocate(16);//attachement
                        //将一个byteBuffer作为附件关联到selectionKey上
                        SelectionKey scKey = sc.register(selector, 0, buffer);
                        scKey.interestOps(SelectionKey.OP_READ);//关注read事件
                        log.debug("{}", sc);
                        log.debug("scKey:{}", scKey);
                    }else if(key.isReadable()){
        
                        try {
        
                            SocketChannel channel = (SocketChannel) key.channel();//拿到触发事件的channel
                            //获取selectionKey的关联附件
                            ByteBuffer buffer = (ByteBuffer) key.attachment();
                            int read = channel.read(buffer);//如果是客户端正常断开， read方法的返回值是-1
                            if(read == -1){
        
                                key.cancel();
                            }else {
        
                                //对数据进行处理，防止粘包半包
                                split(buffer);
                                //需要扩容[一个buffer存放不下一条完整的数据]
                                if(buffer.position() == buffer.limit()){
        
                                    ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
                                    buffer.flip();
                                    newBuffer.put(buffer);// 0123456789abcdef3333\n
                                    key.attach(newBuffer);
                                }
                            }
                        } catch (IOException e){
        
                            e.printStackTrace();
                            key.cancel();//因为客户端断开了，因此需要将key取消（从selector的keys集合中真正删除key）
                        }
                    }
                }
            }
        }
    
        /**
         * 处理粘包半包问题
         * @param source
         */
        private static void split(ByteBuffer source){
        
            //limit = position  position = 0【切换为读模式】
            source.flip();
            for(int i = 0; i < source.limit(); i++){
        
                //根据规定字符找到一条完整消息【'\n'】
                if(source.get(i) == '\n'){
        
                    //fad998877fa\n221  [source.position()当前读取到得位置]
                    int length = i + 1 - source.position();
                    //将这条完整消息存入新的ByteBuffer
                    ByteBuffer target = ByteBuffer.allocate(length);
                    //从source读，向target写
                    for(int j = 0; j < length; j++){
        
                        target.put(source.get());
                    }
                    debugAll(target);
                }
            }
            source.compact(); // 0123456789abcdef  position 16 limit 16
        }
    
    }

Client

@Slf4j
    public class Client {
        
        public static void main(String[] args) throws IOException {
        
            SocketChannel sc = SocketChannel.open();
            sc.connect(new InetSocketAddress("localhost", 8888));
            SocketAddress address = sc.getLocalAddress();
            // sc.write(Charset.defaultCharset().encode("hello\nworld\n"));
            sc.write(Charset.defaultCharset().encode("0123\n456789abcdef"));
            sc.write(Charset.defaultCharset().encode("0123456789abcdef3333\n"));
            System.in.read();
        }
    }

###### \[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 事件

服务端：

public class WriteServer {
        
    
        public static void main(String[] args) throws IOException {
        
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.configureBlocking(false);//设置非阻塞
            ssc.bind(new InetSocketAddress(8888));
    
            Selector selector = Selector.open();
            //注册selector到ssc，同时监听accept事件
            ssc.register(selector, SelectionKey.OP_ACCEPT);
            while(true){
        
                selector.select();
                //拿到绑定到selector上的所有key
                Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                while(iter.hasNext()){
        
                    SelectionKey key = iter.next();
                    iter.remove();
                    if(key.isAcceptable()){
        
                        //ssc只有一个，因此可以直接获取
                        SocketChannel sc = ssc.accept();
                        sc.configureBlocking(false);
                        SelectionKey scKey = sc.register(selector, SelectionKey.OP_READ);
                        //1. 向客户端发送内容
                        StringBuilder sb = new StringBuilder();
                        for (int i = 0; i < 30000000; i++) {
        
                            sb.append("a");
                        }
                        ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());
                        int write = sc.write(buffer);
                        //3. write：实际写了多少字节
                        System.out.println("实际写入字节：" + write);
                        //4. 如果有剩余未读字节，才需要关注写事件
                        if(buffer.hasRemaining()){
        
                            //read：1        write：4
                            //在原有关注事件的基础上 + 额外关注写事件
                            scKey.interestOps(scKey.interestOps() + SelectionKey.OP_WRITE);
                            //把buffer作为附件加入sckey
                            scKey.attach(buffer);
                        }
                    }else if(key.isWritable()){
        
                        //拿到附件
                        ByteBuffer buffer = (ByteBuffer) key.attachment();
                        SocketChannel sc = (SocketChannel) key.channel();
                        int write = sc.write(buffer);
                        System.out.println("实际写入字节数：" + write);
                        if(!buffer.hasRemaining()){
        //没有剩余字节，写完了
                            key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);
                            key.attach(null);
                        }
                    }
                }
            }
        }
    
    }

客户端：

public class WriteClient {
        
        public static void main(String[] args) throws IOException {
        
            Selector selector = Selector.open();
            SocketChannel sc = SocketChannel.open();
            sc.configureBlocking(false);
            sc.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
            sc.connect(new InetSocketAddress("localhost", 8888));
            int count = 0; //统计接收到得数据量
            while(true){
        
                selector.select();
                Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                while(iter.hasNext()){
        
                    SelectionKey key = iter.next();
                    iter.remove();//拿到key之后要remove，否则会重复消费
                    if(key.isConnectable()){
        
                        System.out.println(sc.finishConnect());
                    }else if(key.isReadable()){
        
                        ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
                        count += sc.read(buffer);
                        buffer.clear();
                        System.out.println("总共接收到得数据量：" + count);
                    }
                }
            }
        }
    }

实现多次读写：  
![在这里插入图片描述][3ca2a43a4a79488e887c806250242e30.png]

[1]: https://blog.csdn.net/weixin_45565886/article/details/128528535
[4cd985ec60694743b35c24cccefde768.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/dde16ee6cab94f8e9e8e87ecab6c3f0b.png
[cd64f89aef634fd38575bd84a31721ab.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/4e2ea0809e33466e92dd0f635cc721c7.png
[d796ccf5a1a740dc8b9cd1f49b6f0504.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/a8f17c28153d4bd58669fd6f7338746e.png
[a671f9ddd7594639a938051083c98e63.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/51bae00a60214016a4494a03729a4c50.png
[c5ed6e12242d4af5bfd5f0610caa3729.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/437b2d2accc849348e7cdfebfd8c32ff.png
[2ebb1a801e4b419ba62227e46a197a76.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/0a93cfa6ee384e989ace216e730cf18e.png
[0d11da36ce414224ac2a41c52f61bf30.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/9eaced3e43a0434ab4979a1a07471814.png
[3ca2a43a4a79488e887c806250242e30.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/0c47d7d456bc42e8a8a359fa727694ea.png