Java NIO 学习--Selector

Posted 2022-12-12 智公博客

在之前讲解的网络相关的channel，都有讲到非阻塞模式，只简单说明了那些方法在非阻塞模式下的返回情况，并没有实际的应用；本节要讲到的selector就是NIO中非阻塞模式使用的一大优点；

一、概述

selector，选择器，同过一个选择器，程序可以通过一个线程处理多个channel，而不需要像之前ServerSocketChannel那样每接收一个请求都单开一个线程处理通信；selector基于事件驱动的方式处理多个通道I/O；

二、selector使用

1、创建

一个selector的创建，都是通过简单open静态方法获取：

Selector selector = Selector.open();

2、通道注册

通道要通过selector管理，必须先将通道注册到一个selector上：

serverChannel.configureBlocking(false);
SelectionKey key = serverChannel.register(selector, SelectionKey.OP_ACCEPT);

1.register方法是在SelectableChannel抽象类中定义的，所以只有基础了SelectableChannel类的通道类型才可注册到selector，如ServerSocketChannel、SocketChannel，而且通道必须是设置为非阻塞模式，所以想FileChannel通道，是不能与selector配合使用的；
2.register方法中的第二个参数，表示这个通道注册所感兴趣的事件，总共有4个取值：

connect-连接事件

accept-连接接收事件

read-读事件

write-写事件
如果有多个感兴趣事件，入参可以使用 | 操作

3.register方法的返回值是一个SelectionKey对象，后面再详细讲解这个对象；

3、通过selector选择通道

1. 向一个selector注册完一个或多个通道后，就可以通过三个select方法获取感兴趣事件已就绪的通道：

int select() 阻塞直至有一个注册通道的事件发送

int select(long timeout) 阻塞超时时间为timeout

int selectNow() 不阻塞，立即返回，如没有通道事件发生，返回值为0

select方法返回的int值表示有多少个通道在上一次select后发生了注册感兴趣事件

1. select方法在阻塞期间，如果有其它线程调用了selector的wakeUp方法，正在阻塞的select方法会立即返回，如果wakeUp方法调用时，selector没有select方法在阻塞，那么下次有调用select方法会立即返回；

2. 调用select方法得知有一个或多个通道就绪后，通过selectedKeys方法获取已选择键值(select key set)

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

注册通道时register方法也是会返回一个SelectionKey对象，可以认为该对象包装了对应的通道，可以通过SelectionKey对象获取以下内容：

//获取注册的感兴趣事件集，可以通过
//interestSet & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT；或者
//key.isConnectable(); 判断事件类型
int interestOps = key.interestOps();

//获取已经就绪的事件集，类型判断同上
int readyOps = key.readyOps();

//获取附加对象，该对象可以通过key.attach(obj);方法添加
//也可以在通道注册的时候通过register方法第三个参数带入
//改附加对象可以是通道实用缓存区、用于判断通道的标识等
Object attachment = key.attachment();

//获取这个key所对应的通道对象
SelectableChannel channel2 = key.channel();

//获取这个key所对应的selector
Selector selector = key.selector();

1. 获取到已选择键值(其实是已就绪的通道)，就可以遍历处理这个就绪的集合了，一般方式如下：

Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectedKeys.iterator();
while (iterator.hasNext()) 
    SelectionKey selectionKey = iterator.next();
    if (selectionKey.isAcceptable()) 
        //连接请求时间
     else if (selectionKey.isReadable()) 
        //可读事件 
    
    else if (selectionKey.isConnectable()) 
        //连接事件
    
    else if (selectionKey.isWritable())
        //可写事件
    
    //处理完成后，需要移除
    iterator.remove();

通过4个isXXXXable判断事件类型，并可类型转换为对应的通道类型处理IO事件；
每次循环后需要移除处理完的事件，否则下次selectedKeys()还会再次获取到这事件；

三、实例说明

在ServerSocketChannel与SocketChannel一节的例子中，演示一个简单的网络通信：服务端使用主线程接收请求，每成功接收到一个请求后，创建一个独立的线程处理与客户端的通信；本节使用selector改造这个演示，只使用一个线程处理请求和通信：

客户端：

public class ChannelSelector 

    public static void main(String args[]) throws IOException 
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.bind(new InetSocketAddress(1234));
        serverChannel.configureBlocking(false);

        Selector selector = Selector.open();
        SelectionKey key = serverChannel.register(selector, SelectionKey.OP_ACCEPT);

        while (true) 
            int select = selector.select();
            if (select > 0) 
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectedKeys.iterator();
                while (iterator.hasNext()) 
                    SelectionKey selectionKey = iterator.next();
                    // 接收连接请求
                    if (selectionKey.isAcceptable()) 
                        ServerSocketChannel channel = (ServerSocketChannel) selectionKey
                                .channel();
                        SocketChannel socketChannel = channel.accept();
                        System.out.println("接收到连接请求:"
                                + socketChannel.getRemoteAddress().toString());
                        socketChannel.configureBlocking(false);
                        //每接收请求，注册到同一个selector中处理
                        socketChannel.register(selector, SelectionKey.OP_READ);
                     else if (selectionKey.isReadable()) 
                        // read
                        receiveMessage(selectionKey);
                    
                    iterator.remove();
                
            
        
    

    public static void receiveMessage(SelectionKey selectionKey)
            throws IOException 
        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
        String remoteName = socketChannel.getRemoteAddress().toString();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        ByteBuffer sizeBuffer = ByteBuffer.allocate(4);
        StringBuilder sb = new StringBuilder();
        byte b[];
        try 
            sizeBuffer.clear();
            int read = socketChannel.read(sizeBuffer);
            if (read != -1) 
                sb.setLength(0);
                sizeBuffer.flip();
                int size = sizeBuffer.getInt();
                int readCount = 0;
                b = new byte[1024];
                // 读取已知长度消息内容
                while (readCount < size) 
                    buffer.clear();
                    read = socketChannel.read(buffer);
                    if (read != -1) 
                        readCount += read;
                        buffer.flip();
                        int index = 0;
                        while (buffer.hasRemaining()) 
                            b[index++] = buffer.get();
                            if (index >= b.length) 
                                index = 0;
                                sb.append(new String(b, "UTF-8"));
                            
                        
                        if (index > 0) 
                            sb.append(new String(b, "UTF-8"));
                        
                    
                
                System.out.println(remoteName + ":" + sb.toString());
            
         catch (Exception e) 
            System.out.println(remoteName + " 断线了,连接关闭");
            try 
                //取消这个通道的注册，关闭资源
                selectionKey.cancel();
                socketChannel.close();
             catch (IOException ex) 
            
        
    

服务端还是与之前一样：

public class SocketChanneClient 

    public static void main(String[] args) throws IOException 

        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(new InetSocketAddress(1234));
        while (true) 
            Scanner sc = new Scanner(System.in);
            String next = sc.nextLine();

            sendMessage(socketChannel, next);
        
    

    public static void sendMessage(SocketChannel socketChannel, String mes) throws IOException 
        if (mes == null || mes.isEmpty()) 
            return;
        
        byte[] bytes = mes.getBytes("UTF-8");
        int size = bytes.length;
        ByteBuffer buffer = ByteBuffer.allocate(size);
        ByteBuffer sizeBuffer = ByteBuffer.allocate(4);

        sizeBuffer.putInt(size);
        buffer.put(bytes);

        buffer.flip();
        sizeBuffer.flip();
        ByteBuffer dest[] = sizeBuffer,buffer;
        System.out.println("send message size=" + size + ",content=" + mes);
        while (sizeBuffer.hasRemaining() || buffer.hasRemaining()) 
            socketChannel.write(dest);
        
    

四、selector非阻塞IO的优点

1、阻塞IO的缺点：

1. 当客户端连接多时，需要使用大量线程处理，占用更多的系统资源；
2. 多个线程间的切换许多情况下是无意义的，因为未知阻塞时间；

2、非阻塞IO的优点：

1. 由一个线程来专门处理所有IO事件，并可分发；
2. 基于事件驱动机制，当事件就绪时触发，而不是同步监视事件；