Java NIO - Channel与Selector

声明：

本文由Yasin Shaw原创，首发于我的网站yasinshaw.com和公众号"xy的技术圈"。

若是须要转载请联系我（微信：yasinshaw）并在文章开头显著的地方注明出处。

关注公众号便可获取学习资源或加入技术交流群。

Java NIO有三个核心的组件：Buffer、Channel和Selector。

在上一篇文章中，咱们已经介绍了Buffer，这篇文章主要介绍剩下两个组件：Channel和Selector。

Channel

Channel翻译过来是“通道”的意思，全部的Java NIO都要通过Channel。一个Channel对象其实就对应了一个IO链接。Java NIO中主要有如下Channel实现：

• FileChannel
• DatagramChannel
• SocketChannel
• ServerSocketChannel

分别用于处理文件IO、UDP、TCP客户端、TCP服务端。

这里以ServerSocketChannel和SocketChannel为例，介绍一些经常使用的方法。

// server:
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 8080));
SocketChannel socketChannel = serverSocketChannel.accept();
ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
int readBytes = socketChannel.read(buffer);
if (readBytes > 0) {
    buffer.flip();
    byte[] bytes = new byte[buffer.remaining()];
    buffer.get(bytes);
    String body = new String(bytes, StandardCharsets.UTF_8);
    System.out.println("server 收到：" + body);
}
复制代码

对于服务端来讲，先用open方法建立一个对象，而后使用bind方法绑定端口。

绑定之后，使用accept方法等待新的链接进来，这个方法是阻塞的。一旦有了新的链接，才会解除阻塞。再次调用能够阻塞等待下一个链接。

与Buffer配合，使用read方法能够把数据从Channel读到Buffer里面，而后作后续处理。

// Client:
SocketChannel socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("127.0.0.1", 8080));
ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
buffer.put("hi, 这是client".getBytes(StandardCharsets.UTF_8));
buffer.flip();
socketChannel.write(buffer);
复制代码

对于客户端来讲，有一些微小的区别。客户端不须要bind监听端口，而是直接connect去尝试链接服务端。

一样与Buffer配合，Channel使用write方法能够把数据从Buffer写到Channel里，而后后续就能够作网络传输了。

Selector

Selector翻译过来叫作“选择器”，Selector容许一个线程处理多个Channel。Selector的应用场景是：若是你的应用打开了多个链接（Channel），但每一个链接的流量都很低。好比：聊天服务器或者HTTP服务器。

使用Selector很简单。使用open方法建立一个Selector对象，而后把Channel注册到Selector上。

// 建立一个Selector
Selector selector = Selector.open();

// 把一个Channel注册到Selector
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
复制代码

须要注意的是，必定要使用configureBlocking(false)把Channel设置成非阻塞模式，不然会抛出IllegalBlockingModeException异常。

Channel的register有两个重载方法：

SelectionKey register(Selector sel, int ops) {
    return register(sel, ops, null);
}
SelectionKey register(Selector sel, int ops, Object att);
复制代码

对于ops参数，即selector要关心这个Channel的事件类型，在SelectionKey类里面有这样几个常量：

• OP_READ 能够从Channel读数据
• OP_WRITE 能够写数据到Channel
• OP_CONNECT 链接上了服务器
• OP_ACCEPT 有新的链接进来了

若是你对不止一种事件感兴趣，使用或运算符便可，以下：

int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;
复制代码

须要注意的是，FileChannel只有阻塞模式，不支持非阻塞模式，因此它是没有register方法的！

第三个参数att是attachment的缩写，表明能够传一个“附件”进去。在返回的SelectionKey对象里面，能够获取如下对象：

• channel()：获取Channel
• selector()：获取Selector
• attachment()：获取附件
• attach(obj)：更新附件

除此以外，还有一些判断当前状态的方法：

• isReadable()
• isWritable()
• isConnectable()
• isAcceptable()

通常来讲，咱们不多直接使用单个的SelectionKey，而是从Selector里面轮询全部的SelectionKey，好比：

while (selector.select() > 0) {
     Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
     while(keyIterator.hasNext()) {
         SelectionKey key = keyIterator.next();
         if (key.isReadable()) {
             SocketChannel socketChannel = (SocketChannel) key.channel();
             // read
         } else if(key.isAcceptable()) {
             // accept
         }
         // 其它条件
         keyIterator.remove();
     }
 }
复制代码

Selector能够返回两种SelectionKey集合：

• keys()：已注册的键的集合
• selectedKeys()：已选择的键的集合

并非全部注册过的键都仍然有效，有些可能已经被cancel()方法被调用过的键。因此通常来讲，咱们轮询selectedKeys()方法。

完整的示例代码

如下是一个完整的Server-Client Demo：

Server：

public class Server {
    public static void main(String[] args) {
        try (
                ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
                Selector selector = Selector.open();
        ) {
            serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 8080));
            serverSocketChannel.configureBlocking(false);
            System.out.println("server 启动...");
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            while (selector.select() > 0) {
                Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
                while(keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    if (key.isReadable()) {
                        ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
                        SocketChannel socketChannel = (SocketChannel) key.channel();
                        int readBytes = socketChannel.read(buffer);
                        if (readBytes > 0) {
                            buffer.flip();
                            byte[] bytes = new byte[buffer.remaining()];
                            buffer.get(bytes);
                            String body = new String(bytes, StandardCharsets.UTF_8);
                            System.out.println("server 收到：" + body);
                        }
                    } else if(key.isAcceptable()) {
                        SocketChannel socketChannel = serverSocketChannel.accept();
                        socketChannel.configureBlocking(false);
                        socketChannel.register(selector, SelectionKey.OP_READ);
                    }
                    keyIterator.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
复制代码

Client：

public class Client {
    public static void main(String[] args) {
        try (
                SocketChannel socketChannel = SocketChannel.open();
        ) {
            socketChannel.connect(new InetSocketAddress("127.0.0.1", 8080));
            System.out.println("client 启动...");

            ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
            buffer.put("hi, 这是client".getBytes(StandardCharsets.UTF_8));
            buffer.flip();
            socketChannel.write(buffer);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
复制代码

更多文章，欢迎关注公众号：xy的技术圈