NIO之Channel、Selector

妖狐艹你老母 2022-02-13 11:01 301阅读 0赞

# **Channel** #

JavaNIO Channels和流有一些相似，但是又有些不同：

*  你可以同时读和写Channels，流Stream只支持单向的读或写（InputStream/OutputStream）
 *  Channels可以异步的读和写，流Stream是同步的
 *  Channels总是读取到buffer或者从buffer中写入

下面分别介绍一下Channel最重要的一些实现类：

*  FileChannel : 可以读写文件中的数据
 *  DatagramChannel：可以通过UDP协议读写数据
 *  SocketChannel：可以通过TCP协议读写数据
 *  ServerSocketChannel：允许我们像一个web服务器那样监听TCP链接请求，为每一个链接请求创建一个SocketChannel

下面是一个基本的使用FileChannel读取数据到buffer的例子：

public class FileChannelExam {
 public static void main(String[] args){
 try {
 
 String path = FileChannelExam.class.getResource("/data/nio-data.txt").getPath();
 
 // 创建一个文件通道
 RandomAccessFile file = new RandomAccessFile(path, "rw");
 FileChannel channel = file.getChannel();
 
 // 创建一个字节buffer
 ByteBuffer buffer = ByteBuffer.allocate(1024);
 
 // 读取数据到buffer
 int len = channel.read(buffer);
 
 while (len != -1){
 System.out.println("Read " + len);
 
 // 将写模式转变为读模式，
 // 将写模式下的buffer内容最后位置设为读模式下的limit位置，作为读越界位，同时将读位置设为0
 // 表示转换后重头开始读，同时消除写模式的mark标记
 buffer.flip();
 
 // 判断当前读取位置是否到达越界位（position < limit）
 while (buffer.hasRemaining()){
 // 读取当前position的字节（position++）
 System.out.println(buffer.get());
 }
 
 // 清空当前buffer内容
 buffer.clear();
 len = channel.read(buffer);
 }
 } catch (FileNotFoundException e) {
 e.printStackTrace();
 } catch (IOException e) {
 e.printStackTrace();
 }
 }
 }

需要注意buffer.flip()方法，首先我们从Channel读取数据写入到Buffer，然后调用flip将切换到读模式，才能从buffer中读取数据。

## Channel到Channel的数据传输 ##

在Java NIO中我们可以直接将数据从一个Channel传输到另一个Channel中，比如FileChannel中有transferTo()和transferFrom()方法。

### **transferFrom()** ###

transferFrom()方法可以将一个源channel中的数据传输到一个FileChannel中

String fromPath = FileChannelExam2.class.getResource("/data/nio-data.txt").getPath();
    String toPath = FileChannelExam2.class.getResource("/data/nio-data-to.txt").getPath();
    
    RandomAccessFile fromFile = new RandomAccessFile(fromPath, "rw");
    FileChannel fromChannel = fromFile.getChannel();
    
    RandomAccessFile toFile = new RandomAccessFile(toPath, "rw");
    FileChannel toChannel = toFile.getChannel();
    
    long position = 0;
    long count = fromChannel.size();
    
    toChannel.transferFrom(fromChannel, position, count);

transferFrom()有三个参数，源channel，position，count；position定义目标channel写入的起始位置，count定义写入数据的容量，如果源channel中的数据量小于count，只会写入源channel数据的量。  
另外，在SocketChannel的实现中，当前SocketChannel已经读取一部分数据，稍后仍会读取更多数据情况下，并不一定能将完整的数据读取到FileChannel中。

## transferTo() ##

transferTo()方法可以将FileChannel中的数据传输到其他channel中

String fromPath = FileChannelExam2.class.getResource("/data/nio-data.txt").getPath();
    String toPath = FileChannelExam2.class.getResource("/data/nio-data-to.txt").getPath();
    
    RandomAccessFile fromFile = new RandomAccessFile(fromPath, "rw");
    FileChannel fromChannel = fromFile.getChannel();
    
    RandomAccessFile toFile = new RandomAccessFile(toPath, "rw");
    FileChannel toChannel = toFile.getChannel();
    
    long position = 0;
    long count = fromChannel.size();
    
    fromChannel.transferTo(position, count, toChannel);

上边两个例子有些相似，唯一的区别就是调用的方法和调用方法的对象。这个方法和SocketChannel也会存在和transferFrom同样的问题。

# **Selector** #

Selector是Java NIO中用于管理一个或多个Channel的组件，控制决定对哪些Channel进行读写；通过使用Selector让一个单线程可以管理多个Channel甚至多个网络连接。

使用Selector最大的优势就是可以在较少的线程中控制更多的Channel。事实上我们可以使用一个线程控制需要使用的所有Channel。操作系统线程的运行和切换需要一定的开销，使用的线程越小，系统开销也就越少；因此使用Selector可以节省很多系统开销。下图展示了一个线程使用Selector控制三个Channel的情形。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70][]

### **选择器（Selector)** ###

> Selector选择器类管理着一个被注册的通道集合的信息和它们的就绪状态。通道是和选择器一起被注册的，并且使用选择器来更新通道的就绪状态。当这么做的时候，可以选择将被激发的线程挂起，直到有就绪的的通道。

### **可选择通道(SelectableChannel)** ###

> SelectableChannel这个抽象类提供了实现通道的可选择性所需要的公共方法。它是所有支持就绪检查的通道类的父类。因为FileChannel类没有继承SelectableChannel，因此不是可选通道，而所有socket通道都是可选择的，包括从管道(Pipe)对象的中获得的通道。SelectableChannel可以被注册到Selector对象上，同时可以指定对那个选择器而言，那种操作是感兴趣的。一个通道可以被注册到多个选择器上，但对每个选择器而言只能被注册一次。

*  ServerSocketChannel：应用服务器程序的监听通道。只有通过这个通道，应用程序才能向操作系统注册支持“多路复用IO”的端口监听。同时支持UDP协议和TCP协议。
 *  ScoketChannel：TCP Socket套接字的监听通道，一个Socket套接字对应了一个客户端IP：端口 到 服务器IP：端口的通信连接。
 *  DatagramChannel：UDP 数据报文的监听通道。

### 选择键(SelectionKey) ###

> 选择键封装了特定的通道与特定的选择器的注册关系。选择键对象被SelectableChannel.register()返回并提供一个表示这种注册关系的标记。选择键包含了两个比特集(以整数的形式进行编码)，指示了该注册关系所关心的通道操作，以及通道已经准备好的操作。

JAVA NIO共定义了四种操作类型：OP\_READ、OP\_WRITE、OP\_CONNECT、OP\_ACCEPT（定义在SelectionKey中），分别对应读、写、请求连接、接受连接等网络Socket操作。ServerSocketChannel和SocketChannel可以注册自己感兴趣的操作类型，当对应操作类型的就绪条件满足时OS会通知channel，下表描述各种Channel允许注册的操作类型，Y表示允许注册，N表示不允许注册，其中服务器SocketChannel指由服务器ServerSocketChannel.accept()返回的对象。

<table style="margin-left:4.65pt;"> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:125.7pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:52.25pt;"> OP_READ </td> 
 <td style="vertical-align:top;width:63.8pt;"> OP_WRITE </td> 
 <td style="vertical-align:top;width:63.8pt;"> OP_CONNECT </td> 
 <td style="vertical-align:top;width:85.05pt;"> OP_ACCEPT </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:125.7pt;"> 服务器ServerSocketChannel </td> 
 <td style="vertical-align:top;width:52.25pt;"> N </td> 
 <td style="vertical-align:top;width:63.8pt;"> N </td> 
 <td style="vertical-align:top;width:63.8pt;"> N </td> 
 <td style="vertical-align:top;width:85.05pt;"> Y </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:125.7pt;"> 服务器SocketChannel </td> 
 <td style="vertical-align:top;width:52.25pt;"> Y </td> 
 <td style="vertical-align:top;width:63.8pt;"> Y </td> 
 <td style="vertical-align:top;width:63.8pt;"> N </td> 
 <td style="vertical-align:top;width:85.05pt;"> N </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:125.7pt;"> 客户端SocketChannel </td> 
 <td style="vertical-align:top;width:52.25pt;"> Y </td> 
 <td style="vertical-align:top;width:63.8pt;"> Y </td> 
 <td style="vertical-align:top;width:63.8pt;"> Y </td> 
 <td style="vertical-align:top;width:85.05pt;"> N </td> 
 </tr> 
 </tbody> 
</table>

1.  服务器启动ServerSocketChannel，关注OP\_ACCEPT事件。
2.  客户端启动SocketChannel，连接服务器，关注OP\_CONNECT事件。
3.  服务器接受连接，启动一个服务器的SocketChannel，这个SocketChannel可以关注OP\_READ、OP\_WRITE事件，一般连接建立后会直接关注OP\_READ事件。
4.  客户端这边的客户端SocketChannel发现连接建立后，可以关注OP\_READ、OP\_WRITE事件，一般是需要客户端需要发送数据了才关注OP\_READ事件。
5.  连接建立后客户端与服务器端开始相互发送消息（读写），根据实际情况来关注OP\_READ、OP\_WRITE事件。

### 1.创建Selector ###

Selector selector = Selector.open();

### 2.注册Channel ###

想要通过Selector中控制Channel，必须将Channel注册到Selector中，通过SelectableChannel.register()方法实现。

channel.configureBlocking(false);
    SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

需要注意的是注册到Selector的Channel必须是非阻塞模式的（non-blocking），FileChannel是无法使用的因为FileChannel无法切换到非阻塞模式，SocketChannel非常适合配合Selector使用。

如果在注册Channel的时候希望监听多个事件可以使用“|”连接静态变量

SelectionKey key = channel.register(selector, 
        SelectionKey.OP_READ|SelectionKey.OP_WRITE);

### 3.SelectionKey对象 ###

Channel注册到Selector后会返回一个SelectionKey对象，这个对象包含了下面一些重要属性：

*  事件监听集合（interest set）

监听集合（interest set）是channel在selector监听的事件类型的集合，可以同SelectionKey读写这个配置。

int interestSet = selectionKey.interestOps();
    
    // 通过 & 操作判断监听配置是否包含某类事件
    boolean isInterestedInAccept = interestSet & SelectionKey.OP_ACCEPT;
    boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
    boolean isInterestedInRead = interestSet & SelectionKey.OP_READ;
    boolean isInterestedInWrite = interestSet & SelectionKey.OP_WRITE;

*  就绪结合（ready set）

就绪集合（ready set）是channel已经就绪的操作的集合，我们主要在一个selection操作后访问就绪集合。

int readySet = selectionKey.readyOps();
    // 可以使用和interest set 同样的方法测试集合中是否包含某类事件，
    // 也可以通过调用下边的一些方法进行判断：
    selectionKey.isAcceptable();
    selectionKey.isConnectable();
    selectionKey.isReadable();
    selectionKey.isWritable();

*  Channel对象

Channel channel = selectionKey.channel();

*  Selector对象

Selector selector = selectionKey.selector();

*  一个可选附属对象（an attached object (optional) ）

可以给SelectionKey添加一个附加对象，通常用来标记Channel或者Channel的特征信息。例如，我们可以将和Channel配合使用的Buffer附加到SelectionKey上。

//  附加对象
    selectionKey.attach(theObject);
    // 获取附加对象 
    Object attachedObj = selectionKey.attachment();
    // 还可以再注册channel的时候直接添加附加对象
    SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

### 4.通过Selector选择Channel ###

将多个Channel注册到Selector后，我们就可以通过调用select()方法选择监听了特定事件（connect，accept，read，write）并且已经就绪的Channel。换种说法就是，如果你已经注册了一个监听read事件的channel，它就会通过select()方法接收到read事件。  
select方法有几种不同的重载：

*  int select()：**阻塞**直到至少有一个channel对监听的事件操作准备就绪
 *  int select(long timeout)：和select()方法一样，但只会**阻塞**到指定的超时时间；
 *  int selectNow()：不会**阻塞**，无论是否有就绪的channel都会立即返回。

三个方法的返回值是最后一次调用select()后就绪的channel的数量，如果你调用select()返回1，表示调用select()后有一个channel准备就绪了；当你再次调用sleect()时再返回1，表示这次又有一个channel就绪了，如果对第一次调用就绪的channel没有做任何操作，这时总共有两个已经准备就绪的channel，在两次调用中都只有一个channel变为就绪状态。

## 5.selectionKey() ##

调用select()方法返回就绪channel个数后，可以调用selectedKeys()方法获取就绪channel的SelectionKey集合

Set<SelectionKey> selectedKeys = selector.selectedKeys();

我们可以通过这个集合访问已经就绪的channel

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();
 }

### 6.wakeUp() ###

一个线程调用select()后可以通过再次调用select()离开阻塞状态；也可以通过其他线程调用wakeUp()方法是阻塞在select()的Selecor立即返回。

### 7.close() ###

使用完Selector后可以使用close()方法关闭它，这会关闭Selector和清除注册到Selector的SelecionKey对象，但Channel本身并不会关闭。

### 8.完整流程（伪代码） ###

Selector selector = Selector.open();
 
 channel.configureBlocking(false);
 
 SelectionKey key = channel.register(selector, SelectionKey.OP_READ);
 
 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();
 }
 }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70]: /images/20220213/adc0aa073aa74da0b29d7801631fff05.png