Day388.Selector&Pipe&fileLock文件锁&Path&Files&AsynchronousFileChannel异步通道 -NIO(代码片

Posted 2021-09-28 阿昌喜欢吃黄桃

Selector

一、 Selector 简介

1 、Selector 和 Channel 关系

Selector 一般称 为选择器 ，也可以翻译为 多路复用器。它是 Java NIO 核心组件中的一个，用于检查一个或多个 NIO Channel（通道）的状态是否处于可读、可写。

如此可以实现单线程管理多个 channels,也就是可以管理多个网络链接。

使用 Selector 的好处在于：

使用更少的线程来就可以来处理通道了， 相比使用多个线程，避免了线程上下文切换带来的开销。

---

2、 可选择通道(SelectableChannel)

（1）不是所有的 Channel 都可以被 Selector 复用的。

比方说，FileChannel 就不能被选择器复用。判断一个 Channel 能被 Selector 复用，有一个前提：判断他是否继承了一个抽象类 SelectableChannel。如果继承了 SelectableChannel，则可以被复用，否则不能。

（2）SelectableChannel 类提供了实现通道的可选择性所需要的公共方法。它是所有支持就绪检查的通道类的父类。所有 socket 通道，都继承了 SelectableChannel 类都是可选择的，包括从管道(Pipe)对象的中获得的通道。而FileChannel 类，没有继承 SelectableChannel，因此是不是可选通道。

（3）一个通道可以被注册到多个选择器上，但对每个选择器而言只能被注册一次。通道和选择器之间的关系，使用注册的方式完成。SelectableChannel 可以被注册到Selector 对象上，在注册的时候，需要指定通道的哪些操作，是 Selector 感兴趣的。

---

3 、Channel 注册到 Selector

（1）使用 Channel.register（Selector sel，int ops）方法

将一个通道注册到一个选择器时。第一个参数，指定通道要注册的选择器。第二个参数指定选择器需要查询的通道操作(感兴趣操作)。

（2）可以供选择器查询的通道操作（感兴趣操作），从类型来分，包括以下四种：

• 可读 : SelectionKey.OP_READ
• 可写 : SelectionKey.OP_WRITE
• 连接 : SelectionKey.OP_CONNECT
• 接收 : SelectionKey.OP_ACCEPT

如果 Selector 对通道的多操作类型感兴趣，可以用“位或”操作符来实现：

比如：int key = SelectionKey.OP_READ | SelectionKey.OP_WRITE ;

（3）选择器查询的不是通道的操作，而是通道的某个操作的一种就绪状态。

什么是操作的就绪状态？一旦通道具备完成某个操作的条件，表示该通道的某个操作已经就绪，就可以被 Selector 查询到，程序可以对通道进行对应的操作。比方说，某个SocketChannel 通道可以连接到一个服务器，则处于“连接就绪”(OP_CONNECT)。再比方说，一个 ServerSocketChannel 服务器通道准备好接收新进入的连接，则处于“接收就绪”（OP_ACCEPT）状态。还比方说，一个有数据可读的通道，可以说是“读就绪”(OP_READ)。一个等待写数据的通道可以说是“写就绪”(OP_WRITE)。

---

4、 选择键(SelectionKey)

（1）Channel 注册到后，并且一旦通道处于某种就绪的状态，就可以被选择器查询到。这个工作，使用选择器 Selector 的 select（）方法完成。select 方法的作用，对感兴趣的通道操作，进行就绪状态的查询。

（2）Selector 可以不断的查询 Channel 中发生的操作的就绪状态。并且挑选感兴趣的操作就绪状态。一旦通道有操作的就绪状态达成，并且是 Selector 感兴趣的操作，就会被 Selector 选中，放入选择键集合中。

（3）一个选择键，首先是包含了注册在 Selector 的通道操作的类型，比方说SelectionKey.OP_READ。也包含了特定的通道与特定的选择器之间的注册关系。

开发应用程序是，选择键是编程的关键。NIO 的编程，就是根据对应的选择键，进行不同的业务逻辑处理。

（4）选择键的概念，和事件的概念比较相似。一个选择键类似监听器模式里边的一个事件。由于 Selector 不是事件触发的模式，而是主动去查询的模式，所以不叫事件Event，而是叫 SelectionKey 选择键。

---

二、 Selector 的使用方法

1. Selector 的创建

通过调用 Selector.open()方法创建一个 Selector 对象，如下：

//获取 Selector 选择器
Selector selector = Selector.open();

---

2. 注册 Channel 到 Selector

要实现 Selector 管理 Channel，需要将 channel 注册到相应的 Selector 上

    // 1、获取 Selector 选择器
    Selector selector = Selector.open();
    // 2、获取通道
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    // 3.设置为非阻塞
    serverSocketChannel.configureBlocking(false);
    // 4、绑定连接
    serverSocketChannel.bind(new InetSocketAddress(9999));
    // 5、将通道注册到选择器上,并制定监听事件为：“接收”事件
    serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);

上面通过调用通道的 register()方法会将它注册到一个选择器上。

首先需要注意的是：

• （1）与 Selector 一起使用时，Channel 必须处于非阻塞模式下，否则将抛出异常IllegalBlockingModeException。这意味着，FileChannel 不能与 Selector 一起使用，因为 FileChannel 不能切换到非阻塞模式，而套接字相关的所有的通道都可以。
• （2）一个通道，并没有一定要支持所有的四种操作。比如服务器通道ServerSocketChannel 支持 Accept 接受操作，而 SocketChannel 客户端通道则不支持。可以通过通道上的 validOps()方法，来获取特定通道下所有支持的操作集合。

---

3. 轮询查询就绪操作

（1）通过 Selector 的 select（）方法，可以查询出已经就绪的通道操作，这些就绪的状态集合，包存在一个元素是 SelectionKey 对象的 Set 集合中。

（2）下面是 Selector 几个重载的查询 select()方法：

• select():阻塞到至少有一个通道在你注册的事件上就绪了。
• select(long timeout)：和 select()一样，但最长阻塞事件为 timeout 毫秒。
• selectNow():非阻塞，只要有通道就绪就立刻返回。

select()方法返回的 int 值，表示有多少通道已经就绪，更准确的说，是自前一次 select方法以来到这一次 select 方法之间的时间段上，有多少通道变成就绪状态。

例如：

首次调用 select()方法，如果有一个通道变成就绪状态，返回了 1，若再次调用select()方法，如果另一个通道就绪了，它会再次返回 1。如果对第一个就绪的channel 没有做任何操作，现在就有两个就绪的通道，但在每次 select()方法调用之间，只有一个通道就绪了。

一旦调用 select()方法，并且返回值不为 0 时，在 Selector 中有一个 selectedKeys()方法，用来访问已选择键集合，迭代集合的每一个选择键元素，根据就绪操作的类型，完成对应的操作：

//获取已经就绪的通道操作
Set selectedKeys = selector.selectedKeys();
Iterator keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()) {
    SelectionKey key = keyIterator.next();
    //判断key就绪状态是什么
    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();
}

---

5. 停止选择的方法

选择器执行选择的过程，系统底层会依次询问每个通道是否已经就绪，这个过程可能会造成调用线程进入阻塞状态,那么我们有以下三种方式可以唤醒在 select（）方法中阻塞的线程。

wakeup()方法 ：

通过调用 Selector 对象的 wakeup（）方法让处在阻塞状态的select()方法立刻返回(唤醒操作)

该方法使得选择器上的第一个还没有返回的选择操作立即返回。如果当前没有进行中的选择操作，那么下一次对 select()方法的一次调用将立即返回。

close()方法 ：

通过 close（）方法关闭 Selector，该方法使得任何一个在选择操作中阻塞的线程都被唤醒（类似 wakeup（）），

同时使得注册到该 Selector 的所有 Channel 被注销，所有的键将被取消，但是 Channel本身并不会关闭。

---

三、 示例代码

1、服务端代码

@Test
public void ServerDemo() {
    try {
        //获取服务端通道
        ServerSocketChannel ssc = ServerSocketChannel.open();
        //绑定服务端ip端口号
        ssc.socket().bind(new InetSocketAddress("127.0.0.1",8000));
        //切换非阻塞模式
        ssc.configureBlocking(false);
        
        //获取selector选择器
        Selector selector = Selector.open();
        // 将服务端channel注册到选择器上，并且指定感兴趣的事件是 Accept
        ssc.register(selector, SelectionKey.OP_ACCEPT);
        
        //创建读/写缓冲区
        ByteBuffer readBuff = ByteBuffer.allocate (1024);
        ByteBuffer writeBuff = ByteBuffer.allocate (128);
        
        //写入数据
        writeBuff.put("received".getBytes());
        //切换读写模式
        writeBuff.flip();
        
        while (true) {
            int nReady = selector.select();
            //获取就绪状态集合
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> it = keys.iterator();
            while (it.hasNext()) {
                //获取到就绪状态
                SelectionKey key = it.next();
                it.remove();
                
                //判断是什么状态，对对应操作进行对应处理
                if (key.isAcceptable()) {
                    // 创建新的连接，并且把连接注册到 selector 上，而且，声明这个 channel 只对读操作感兴趣。
                    SocketChannel socketChannel = ssc.accept();
                    socketChannel.configureBlocking(false);
                    socketChannel.register(selector, SelectionKey.OP_READ);
                }
                else if (key.isReadable()) {
                    SocketChannel socketChannel = (SocketChannel) key.channel();
                    readBuff.clear();
                    socketChannel.read(readBuff);
                    readBuff.flip();
                    System.out.println("received : " + new String(readBuff.array()));
                    key.interestOps(SelectionKey.OP_WRITE);
                }
                else if (key.isWritable()) {
                    writeBuff.rewind();
                    SocketChannel socketChannel = (SocketChannel) key.channel();
                    socketChannel.write(writeBuff);
                    key.interestOps(SelectionKey. OP_READ );
                }
                
            }
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}

2、客户端代码

@Test
public void ClientDemo() {
    try {
        //获取通道
        SocketChannel socketChannel = SocketChannel.open();
        //绑定主机的ip端口号
        socketChannel.connect(new InetSocketAddress("127.0.0.1",8000));
        //设置非阻塞模式
        socketChannel.configureBlocking(false);
        //创建buffer
        ByteBuffer writeBuffer = ByteBuffer.allocate(32);
        ByteBuffer readBuffer = ByteBuffer.allocate(32);
        //给buffer写入数据
        writeBuffer.put("hello".getBytes());
        //模式切换
        writeBuffer.flip();
        while (true) {
            writeBuffer.rewind();
            //写入通道数据
            socketChannel.write(writeBuffer);
            //关闭
            readBuffer.clear();
            socketChannel.read(readBuffer);
        }
    } catch (IOException e) {

    }
}

---

3 、NIO 编程步骤总结

• 第一步：创建 Selector 选择器
• 第二步：创建 ServerSocketChannel 通道，并绑定监听端口
• 第三步：设置 Channel 通道是非阻塞模式
• 第四步：把 Channel 注册到 Socketor 选择器上，监听连接事件
• 第五步：调用 Selector 的 select 方法（循环调用），监测通道的就绪状况
• 第六步：调用 selectKeys 方法获取就绪 channel 集合
• 第七步：遍历就绪 channel 集合，判断就绪事件类型，实现具体的业务操作
• 第八步：根据业务，决定是否需要再次注册监听事件，重复执行第三步操作

---

Pipe & FileLock

一、 Pipe

Java NIO 管道是 2 个线程之间的单向数据连接。Pipe 有一个 source 通道和一个sink 通道。数据会被写到 sink 通道，从 source 通道读取。

1、创建管道

通过 Pipe.open()方法打开管道。

Pipe pipe = Pipe.open();

---

2、写入管道

要向管道写数据，需要访问 sink 通道。

Pipe.SinkChannel sinkChannel = pipe.sink();

通过调用 SinkChannel 的 write()方法，将数据写入 SinkChannel：

String newData = "New String to write to file..." + System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while(buf.hasRemaining()) {
	sinkChannel.write(buf);
}

---

3、从管道读取数据

从读取管道的数据，需要访问 source 通道，像这样：

Pipe.SourceChannel sourceChannel = pipe.source();

调用 source 通道的 read()方法来读取数据：

ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = sourceChannel.read(buf);

read()方法返回的 int 值会告诉我们多少字节被读进了缓冲区。

---

4、示例

    @Test
    public void testPipe() throws IOException {
    // 1、获取通道
    Pipe pipe = Pipe.open();
    // 2、获取 sink 管道，用来传送数据,写管道
    Pipe.SinkChannel sinkChannel = pipe.sink();
    // 3、申请一定大小的缓冲区
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    byteBuffer.put("achang".getBytes());
    byteBuffer.flip();
        
    // 4、sink 发送数据,写入操作
    sinkChannel.write(byteBuffer);
        
    // 5、创建接收 pipe 数据的 source 管道，读管道
    Pipe.SourceChannel sourceChannel = pipe.source();
        
    // 6、接收数据，并保存到缓冲区中
    ByteBuffer byteBuffer2 = ByteBuffer.allocate(1024);
    int length = sourceChannel.read(byteBuffer2);
    System.out.println(new String(byteBuffer2.array(), 0, length));
        
    // 7、关闭
    sourceChannel.close();
    sinkChannel.close();
    }

---

二、 FileLock文件锁

1、FileLock 简介

文件锁在 OS 中很常见，如果多个程序同时访问、修改同一个文件，很容易因为文件数据不同步而出现问题。给文件加一个锁，同一时间，只能有一个程序修改此文件，或者程序都只能读此文件，这就解决了同步问题。

文件锁是进程级别的，不是线程级别的。文件锁可以解决多个进程并发访问、修改同一个文件的问题，但不能解决多线程并发访问、修改同一文件的问题。使用文件锁时，同一进程内的多个线程，可以同时访问、修改此文件。

文件锁是当前程序所属的 JVM 实例持有的，一旦获取到文件锁（对文件加锁），要调用 release()，或者关闭对应的 FileChannel 对象，或者当前 JVM 退出，才会释放这个锁。

一旦某个进程（比如说 JVM 实例）对某个文件加锁，则在释放这个锁之前，此进程不能再对此文件加锁，就是说 JVM 实例在同一文件上的文件锁是不重叠的（进程级别不能重复在同一文件上获取锁）。

---

2、文件锁分类：

• 排它锁：
  • 又叫独占锁。对文件加排它锁后，该进程可以对此文件进行读写，该进程独占此文件，其他进程不能读写此文件，直到该进程释放文件锁。
• 共享锁：
  • 某个进程对文件加共享锁，其他进程也可以访问此文件，但这些进程都只能读此文件，不能写。线程是安全的。只要还有一个进程持有共享锁，此文件就只能读，不能写。

---

3、使用示例

    //创建 FileChannel 对象，文件锁只能通过 FileChannel 对象来使用
    FileChannel fileChannel=new FileOutputStream("./1.txt").getChannel();
    //对文件加锁
    FileLock lock=fileChannel.lock();
    //对此文件进行一些读写操作。
    //.......
    //释放锁
    lock.release();

文件锁要通过 FileChannel 对象使用。

---

4、获取文件锁方法

• 有 4 种获取文件锁的方法：

lock() ：

//对整个文件加锁，默认为排它锁。
lock(long position, long size, booean shared) //自定义加锁方式。前 2 个参数指定要加锁的部分（可以只对此文件的部分内容加锁），第三个参数值指定是否是共享锁。

tryLock() ：

//对整个文件加锁，默认为排它锁。

tryLock(long position, long size, booean shared)：

//自定义加锁方式。
如果指定为共享锁，则其它进程可读此文件，所有进程均不能写此文件，如果某进程试图对此文件进行写操作，会抛出异常。

---

5、lock 与 tryLock 的区别：

• lock 是阻塞式的，如果未获取到文件锁，会一直阻塞当前线程，直到获取文件锁tryLock 和 lock 的作用相同，只不过 tryLock 是非阻塞式的，

• tryLock 是尝试获取文件锁，获取成功就返回锁对象，否则返回 null，不会阻塞当前线程。

---

6、FileLock 两个方法：

boolean isShared() //此文件锁是否是共享锁
boolean isValid() //此文件锁是否还有效

在某些 OS 上，对某个文件加锁后，不能对此文件使用通道映射。

---

7、完整例子

public class Demo1 {
    public static void main(String[] args) throws IOException {
        String input = "achang";
        System.out.println("输入 :" + input);
        
        ByteBuffer buf = ByteBuffer.wrap(input.getBytes());
        String filePath = "D:\\\\achang\\\\01.txt";
        Path pt = Paths.get(filePath);
        FileChannel channel = FileChannel.open(pt,
                                                 StandardOpenOption.WRITE,
                                                 StandardOpenOption.APPEND);
        
        channel.position(channel.size() - 1); // position of a cursor at the end offile
        
        //文件获取锁
        // 获得锁方法一：lock()，阻塞方法，当文件锁不可用时，当前进程会被挂起
        //lock = channel.lock();// 无参 lock()为独占锁
        // lock = channel.lock(0L, Long.MAX_VALUE, true);有参 lock()为共享锁，有写操作会报异常
        // 获得锁方法二：trylock()，非阻塞的方法，当文件锁不可用时，tryLock()会得到 null 值
        FileLock lock = channel.tryLock(0,Long. MAX_VALUE ,false);
        System.out.println("共享锁 shared: " + lock.isShared());
        channel.write(buf);
        
        channel.close(); // Releases the Lock
        System.out.println("写操作完成.");
        
        //读取数据
        readPrint (fp);
    }

//读取数据
public static void readPrint(String path) throws IOException {
    //通过字符流读取
    FileReader filereader = new FileReader(path);
    //包装缓存流
    BufferedReader bufferedreader = new BufferedReader(filereader);
    
    String tr = bufferedreader.readLine();
    System.out.println("读取内容: ");
    while (tr != null) {
        System.out.println(" " + tr);
        tr = bufferedreader.readLine();
    }
    
    //关闭
    filereader.close();
    bufferedreader.close();
}

---

Path&Files&AsynchronousFileChannel异步通道

一、 Path

1、Path 简介

Java Path 接口是 Java NIO 更新的一部分，同 Java NIO 一起已经包括在 Java6 和Java7 中。Java Path 接口是在 Java7 中添加到 Java NIO 的。

Path 接口位于java.nio.file 包中，所以 Path 接口的完全限定名称为 java.nio.file.Path。Java Path 实例表示文件系统中的路径。一个路径可以指向一个文件或一个目录。

路径可以是绝对路径，也可以是相对路径。绝对路径包含从文件系统的根目录到它指向的文件或目录的完整路径。相对路径包含相对于其他路径的文件或目录的路径。

在许多方面，java.nio.file.Path 接口类似于java.io.File 类，但是有一些差别。不过，在许多情况下，可以使用 Path 接口来替换 File 类的使用。

---

2、创建 Path 实例

使用 java.nio.file.Path 实例必须创建一个Path实例。

可以使

以上是关于Day388.Selector&Pipe&fileLock文件锁&Path&Files&AsynchronousFileChannel异步通道 -NIO(代码片的主要内容，如果未能解决你的问题，请参考以下文章

day26粘包

Can't open named pipe to host: . pipe: MySQL

understanding of Pipe line & Timing Logic

Java Day 22

gnu 并行：结合使用 --pipe 和 args

python 学习笔记 - Queue & Pipes，进程间通讯