zoukankan      html  css  js  c++  java
  • NIO的Buffer&Channel&Selector

    java的NIO和AIO

    Buffer

    • position、limit、capacity
    • 初始化 Buffer
    • 填充 Buffer
    • 提取 Buffer 中的值
    • mark() & reset()
    • rewind() & clear() & compact()

    Channel

    • FileChannel
    • SocketChannel
    • ServerSocketChannel
    • DatagramChannel

    Selector

    Buffer

    一个 Buffer 本质上是内存中的一块,我们可以将数据写入这块内存,之后从这块内存获取数据。

    java.nio 定义了以下几个 Buffer 的实现,其实核心是最后的 ByteBuffer,其他实现类只是包装了一下它而已,我们使用最多的通常也是 ByteBuffer。

    可以将 Buffer 理解为一个数组,IntBuffer、CharBuffer、DoubleBuffer 等分别对应 int[]、char[]、double[] 等。

    position、limit、capacity

    Buffer 中也有几个重要属性:position、limit、capacity。

    capacity,它代表这个缓冲区的容量,一旦设定就不可以更改。比如 capacity 为 1024 的 IntBuffer,代表其一次可以存放 1024 个 int 类型的值。一旦 Buffer 的容量达到 capacity,需要清空 Buffer,才能重新写入值。

    position 和 limit 是变化的,我们分别看下读和写操作下,它们是如何变化的。

    position 的初始值是 0,每往 Buffer 中写入一个值,position 就自动加 1,代表下一次的写入位置。读操作的时候也是类似的,每读一个值,position 就自动加 1。

    写操作模式到读操作模式切换的时候(flip),position 都会归零,这样就可以从头开始读写了。

    limit:写操作模式下,limit 代表的是最大能写入的数据,这个时候 limit 等于 capacity。写结束后,切换到读模式,此时的 limit 等于 Buffer 中实际的数据大小,因为 Buffer 不一定被写满了。

    切换模式

    调用 Buffer 的 flip() 方法,可以进行模式切换。其实这个方法也就是设置了一下 position 和 limit 值。

    public final Buffer flip() {
        limit = position; // 将 limit 设置为实际写入的数据数量
        position = 0; // 重置 position 为 0
        mark = -1; // mark 之后再说
        return this;
    }

    初始化 Buffer

    每个 Buffer 实现类都提供了一个静态方法 allocate(int capacity) 帮助我们快速实例化一个 Buffer。另外经常使用 wrap 方法来初始化一个 Buffer。

    ByteBuffer byteBuf = ByteBuffer.allocate(1024);
    IntBuffer intBuf = IntBuffer.allocate(1024);
    LongBuffer longBuf = LongBuffer.allocate(1024);
    public static ByteBuffer wrap(byte[] array) {
        ...
    }
    

    填充 Buffer

    put 方法:各个 Buffer 类都提供了一些 put 方法用于将数据填充到 Buffer 中,这些方法需要自己控制 Buffer 大小,不能超过 capacity,超过会抛 java.nio.BufferOverflowException 异常。

    通过Channel填充:int num = channel.read(buf);将 Channel 的数据填充到 Buffer 中。在系统层面上,这个操作我们称为读操作,因为数据是从外部(文件或网络等)读到内存中。

    提取 Buffer 中的值

    get 方法:读操作提供了一系列的 get 方法。

    通过Channel读取:int num = channel.write(buf);将buf的内容写入channel。

    mark() & reset()

    mark 用于临时保存 position 的值,每次调用 mark() 方法都会将 mark 设值为当前的 position,便于后续需要的时候使用。

    考虑以下场景,我们在 position 为 5 的时候,先 mark() 一下,然后继续往下读,读到第 10 的时候,我想重新回到 position 为 5 的地方重新来一遍,那只要调一下 reset() 方法,position 就回到 5 了。

    rewind() & clear() & compact()

    rewind():会重置 position 为 0,通常用于重新从头读写 Buffer。

    clear():有点重置 Buffer 的意思,相当于重新实例化了一样。不会将 Buffer 中的数据清空,只不过后续的写入会覆盖掉原来的数据,也就相当于清空了数据了。

    compact():和 clear() 一样的是,它们都是在准备往 Buffer 填充新的数据之前调用。先处理还没有读取的数据,也就是 position 到 limit 之间的数据(还没有读过的数据),先将这些数据移到左边,然后在这个基础上再开始写入。很明显写入模式下,此时 limit 还是等于 capacity,position 指向原来数据的右边。

        public final Buffer clear() {
            position = 0;
            limit = capacity;
            mark = -1;
            return this;
        }
    
        public ByteBuffer compact() {
    
            int pos = position();
            int lim = limit();
            assert (pos <= lim);
            int rem = (pos <= lim ? lim - pos : 0);  //position 到 limit 之间的数据
    
            unsafe.copyMemory(ix(pos), ix(0), (long)rem << 0); //移到左边
            position(rem);
            limit(capacity());
            discardMark();
            return this;
    
        }
    

    Channel

    所有的 NIO 操作始于通道,通道是数据来源或数据写入的目的地。

    • FileChannel:文件通道,用于文件的读和写
    • DatagramChannel:用于 UDP 连接的接收和发送
    • SocketChannel:把它理解为 TCP 连接通道,简单理解就是 TCP 客户端
    • ServerSocketChannel:TCP 对应的服务端,用于监听某个端口进来的请求

    Channel 经常翻译为通道,类似 IO 中的流,用于读取和写入。它与前面介绍的 Buffer 打交道。

    读操作的时候将 Channel 中的数据填充到 Buffer 中,而写操作时将 Buffer 中的数据写入到 Channel 中。

    FileChannel

    FileChannel 是不支持非阻塞的。

    // 读取
    FileInputStream inputStream = new FileInputStream(new File("/data.txt"));
    FileChannel fileChannel = inputStream.getChannel();
    
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    int num = fileChannel.read(buffer);
    
    // 写入
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    buffer.put("随机写入一些内容到 Buffer 中".getBytes());
    // Buffer 切换为读模式
    buffer.flip();
    while(buffer.hasRemaining()) {
        // 将 Buffer 中的内容写入文件
        fileChannel.write(buffer);
    }
    

    SocketChannel

    SocketChannel 是 TCP 客户端,它代表的是一个网络通道,可读可写。

    //打开一个 TCP 连接:
    // 打开一个通道
    SocketChannel socketChannel = SocketChannel.open();
    // 发起连接
    socketChannel.connect(new InetSocketAddress("https://www.baidu.com", 80));
    
    // 读取数据
    socketChannel.read(buffer);
    
    // 写入数据到网络连接中
    while(buffer.hasRemaining()) {
        socketChannel.write(buffer);   
    }
    

    ServerSocketChannel

    ServerSocketChannel 就是对应的服务端。ServerSocketChannel 不和 Buffer 打交道了,因为它并不实际处理数据,它一旦接收到请求后,实例化 SocketChannel,之后在这个连接通道上的数据传递它就不管了,因为它需要继续监听端口,等待下一个连接。

    ServerSocketChannel 用于监听机器端口,管理从这个端口进来的 TCP 连接。

    // 实例化
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    // 监听 8080 端口
    serverSocketChannel.socket().bind(new InetSocketAddress(8080));
    
    while (true) {
        // 一旦有一个 TCP 连接进来,就对应创建一个 SocketChannel 进行处理
        SocketChannel socketChannel = serverSocketChannel.accept();
    }
    

    DatagramChannel

    UDP 和 TCP 不一样,DatagramChannel 一个类处理了服务端和客户端。

    UDP 是面向无连接的,不需要和对方握手,不需要通知对方,就可以直接将数据包投出去,至于能不能送达,它是不知道的。

    //监听端口
    DatagramChannel channel = DatagramChannel.open();
    channel.socket().bind(new InetSocketAddress(9090));
    
    ByteBuffer buf = ByteBuffer.allocate(48);
    buf.clear();
    
    channel.receive(buf);
    
    //发送数据
    String newData = "New String to write to file..."
                        + System.currentTimeMillis();
    
    ByteBuffer buf = ByteBuffer.allocate(48);
    buf.clear();
    buf.put(newData.getBytes());
    buf.flip();
    
    int bytesSent = channel.send(buf, new InetSocketAddress("localhost.com", 80));
    

    Selector

    Selector 建立在非阻塞的基础之上,多路复用指的就是它,用于实现一个线程管理多个 Channel

    //开启一个 Selector  多路复用器
    Selector selector = Selector.open();
    
    //将 Channel 注册到 Selector 上, 注册到 Selector 的 Channel 必须要支持非阻塞模式.FileChannel 不支持非阻塞,最常见的 SocketChannel 和 ServerSocketChannel。
    // 将通道设置为非阻塞模式,因为默认都是阻塞模式的
    channel.configureBlocking(false);
    // 注册
    SelectionKey key = channel.register(selector, SelectionKey.OP_READ);
    //注册方法返回值是 SelectionKey 实例,它包含了 Channel 和 Selector 信息,也包括了一个叫做 Interest Set 的信息,即我们设置的我们感兴趣的正在监听的事件集合。
    //监听感兴趣的事件,可以同时监听一个 Channel 中的发生的多个事件,共以下四种事件:
    //        SelectionKey.OP_READ
    //        SelectionKey.OP_WRITE
    //        SelectionKey.OP_CONNECT
    //        SelectionKey.OP_ACCEPT
    
    //调用 select() 方法获取通道信息。用于判断是否有我们感兴趣的事件已经发生了。
    while(true) {
      // 判断是否有事件准备好
      int readyChannels = selector.select();
      if(readyChannels == 0) continue;
    
      // 遍历
      Set<SelectionKey> selectedKeys = selector.selectedKeys();
      Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
      while(keyIterator.hasNext()) {
        SelectionKey key = keyIterator.next();
    
        if(key.isAcceptable()) {
            // a connection was accepted by a ServerSocketChannel.
    
        } else if (key.isConnectable()) {
            // a connection was established with a remote server.
    
        } else if (key.isReadable()) {
            // a channel is ready for reading
    
        } else if (key.isWritable()) {
            // a channel is ready for writing
        }
    
        keyIterator.remove();
      }
    }
    

      

  • 相关阅读:
    LeetCode 242. Valid Anagram (验证变位词)
    LeetCode 205. Isomorphic Strings (同构字符串)
    LeetCode 204. Count Primes (质数的个数)
    LeetCode 202. Happy Number (快乐数字)
    LeetCode 170. Two Sum III
    LeetCode 136. Single Number (落单的数)
    LeetCode 697. Degree of an Array (数组的度)
    LeetCode 695. Max Area of Island (岛的最大区域)
    Spark中的键值对操作
    各种排序算法总结
  • 原文地址:https://www.cnblogs.com/wade-luffy/p/8448655.html
Copyright © 2011-2022 走看看