Java I/O(4):AIO和NIO中的Selector

时间:2022-10-21 10:08:46

您好,我是湘王,这是我的博客园,欢迎您来,欢迎您再来~

 

Java NIO的三大核心中,除了Channel和Buffer,剩下的就是Selector了。有的地方叫它选择器,也有叫多路复用器的(比如Netty)。

之前提过,数据总是从Channel读取到Buffer,或者从Buffer写入到Channel,单个线程可以监听多个Channel——Selector就是这个线程背后的实现机制(所以得名Selector)。

Java I/O(4):AIO和NIO中的Selector

 

 

 

Selector通过控制单个线程处理多个Channel,如果应用打开了多个Channel,但每次传输的流量都很低,使用Selector就会很方便(至于为什么,具体到Netty中再分析)。所以使用Selector的好处就显而易见:用最少的资源实现最多的操作,避免了线程切换带来的开销。

还是以代码为例来演示Selector的作用。新建一个类,main()方法中输入下面的代码:

/**
 * NIO中的Selector
 *
 * @author xiangwang
 */
public class TestSelector {
    public static void main(String args[]) throws IOException {
        // 创建ServerSocketChannel
        ServerSocketChannel channel1 = ServerSocketChannel.open();
        channel1.socket().bind(new InetSocketAddress("127.0.0.1", 8080));
        channel1.configureBlocking(false);
        ServerSocketChannel channel2 = ServerSocketChannel.open();
        channel2.socket().bind(new InetSocketAddress("127.0.0.1", 9090));
        channel2.configureBlocking(false);

        // 创建一个Selector对象
        Selector selector = Selector.open();
        // 按照字面意思理解,应该是这样的:selector.register(channel, event);
        // 但其实是这样的:channel.register(selector, SelectionKey.OP_READ);
        // 四种监听事件:
        // OP_CONNECT(连接就绪)
        // OP_ACCEPT(接收就绪)
        // OP_READ(读就绪)
        // OP_WRITE(写就绪)
        // 注册Channel到Selector,事件一旦被触发,监听随之结束
        SelectionKey key1 = channel1.register(selector, SelectionKey.OP_ACCEPT);
        SelectionKey key2 = channel2.register(selector, SelectionKey.OP_ACCEPT);

        // 模板代码:在编写程序时,大多数时间都是在模板代码中添加相应的业务代码
        while(true) {
            int readyNum = selector.select();
            if (readyNum == 0) {
                continue;
            }

            Set<SelectionKey> selectedKeys = selector.selectedKeys();
            // 轮询
            for (SelectionKey key : selectedKeys) {
                Channel channel = key.channel();
                if (key.isConnectable()) {
                    if (channel == channel1) {
                        System.out.println("channel1连接就绪");
                    } else {
                        System.out.println("channel2连接就绪");
                    }
                } else if (key.isAcceptable()) {
                    if (channel == channel1) {
                        System.out.println("channel1接收就绪");
                    } else {
                        System.out.println("channel2接收就绪");
                    }
                }
                // 触发后删除,这里不删
                // it.remove();
            }
        }
    }
}

 

 

代码写好后启动ServerSocketChannel服务,可以看到我这里已经启动成功:

Java I/O(4):AIO和NIO中的Selector

 

 

 

然后在网上下载一个叫做SocketTest.jar的工具(在一些工具网站下载的时候当心中毒,如果不放心,可以私信我,给你地址),双击打开,并按下图方式执行:

Java I/O(4):AIO和NIO中的Selector

 

 

 

点击「Connect」可以看到变化:

Java I/O(4):AIO和NIO中的Selector

 

 

 

然后点击「Disconnect」,再输入9090」后,再点击「Connect」试试:

Java I/O(4):AIO和NIO中的Selector

 

 

 

可以看到结果显示结果变了:

Java I/O(4):AIO和NIO中的Selector

 

 

 

两次连接,打印了三条信息:说明selector的轮询在起作用(因为Set<SelectionKey>中包含了所有处于监听的SelectionKey)。但是「接收就绪」监听事件仅执行了一次就再不响应。如果感兴趣的话你可以把OP_READ、OP_WRITE这些事件也执行一下试试看。

 

因为Selector是单线程轮询监听多个Channel,那么如果Selector(线程)之间需要传递数据,怎么办呢?——Pipe登场了。Pipe就是一种用于Selector之间数据传递的「管道」。

先来看个图:

Java I/O(4):AIO和NIO中的Selector

 

 

 

可以清楚地看到它的工作方式。

还是用代码来解释。

/**
 * NIO中的Pipe
 *
 * @author xiangwang
 */
public class TestPipe {
    public static void main(String args[]) throws IOException {
        // 打开管道
        Pipe pipe = Pipe.open();

        // 将Buffer数据写入到管道
        Pipe.SinkChannel sinkChannel = pipe.sink();
        ByteBuffer buffer = ByteBuffer.allocate(32);
        buffer.put("ByteBuffer".getBytes());
        // 切换到写模式
        buffer.flip();
        sinkChannel.write(buffer);

        // 从管道读取数据
        Pipe.SourceChannel sourceChannel = pipe.source();
        buffer = ByteBuffer.allocate(32);
        sourceChannel.read(buffer);
        System.out.println(new String(buffer.array()));

        // 关闭管道
        sinkChannel.close();
        sourceChannel.close();
    }
}

 

 

之前说过,同步指的按顺序一次完成一个任务,直到前一个任务完成并有了结果以后,才能再执行后面的任务。而异步指的是前一个任务结束后,并不等待任务结果,而是继续执行后一个任务,在所有任务都「执行」完后,通过任务的回调函数去获得结果。所以异步使得应用性能有了极大的提高。为了更加生动地说明什么是异步,可以来做个实验:

Java I/O(4):AIO和NIO中的Selector

 

 

 

通过调用CompletableFuture.supplyAsync()方法可以很明显地观察到,处于位置2的「这一步先执行」会最先显示,然后才执行位置1的代码。而这就是异步的具体实现。

NIO为了支持异步,升级到了NIO2,也就是AIO。而AIO引入了新的异步Channel的概念,并提供了异步FileChannel和异步SocketChannel的实现。AIO的异步SocketChannel是真正的异步非阻塞I/O。通过代码可以更好地说明:

/**
 * AIO客户端
 *
 * @author xiangwang
 */
public class AioClient {
    public void start() throws IOException, InterruptedException {
        AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();
        if (channel.isOpen()) {
            // socket接收缓冲区recbuf大小
            channel.setOption(StandardSocketOptions.SO_RCVBUF, 128 * 1024);
            // socket发送缓冲区recbuf大小
            channel.setOption(StandardSocketOptions.SO_SNDBUF, 128 * 1024);
            // 保持长连接状态
            channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
            // 连接到服务端
            channel.connect(new InetSocketAddress(8080), null,
                    new AioClientHandler(channel));
            // 阻塞主进程
            for(;;) {
                TimeUnit.SECONDS.sleep(1);
            }
        } else {
            throw new RuntimeException("Channel not opened!");
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        new AioClient().start();
    }
}

 

/**
 * AIO客户端CompletionHandler
 *
 * @author xiangwang
 */
public class AioClientHandler implements CompletionHandler<Void, AioClient> {
    private final AsynchronousSocketChannel channel;
    private final CharsetDecoder decoder = Charset.defaultCharset().newDecoder();
    private final BufferedReader input = new BufferedReader(new InputStreamReader(System.in));

    public AioClientHandler(AsynchronousSocketChannel channel) {
        this.channel = channel;
    }
    @Override
    public void failed(Throwable exc, AioClient attachment) {
        throw new RuntimeException("channel not opened!");
    }
    @Override
    public void completed(Void result, AioClient attachment) {
        System.out.println("send message to server: ");
        try {
            // 将输入内容写到buffer
            String line = input.readLine();
            channel.write(ByteBuffer.wrap(line.getBytes()));
            // 在操作系统中的Java本地方法native已经把数据写到了buffer中
            // 这里只需要一个缓冲区能接收就行了
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            while (channel.read(buffer).get() != -1) {
                buffer.flip();
                System.out.println("from server: " + decoder.decode(buffer).toString());
                if (buffer.hasRemaining()) {
                    buffer.compact();
                } else {
                    buffer.clear();
                }
                // 将输入内容写到buffer
                line = input.readLine();
                channel.write(ByteBuffer.wrap(line.getBytes()));
            }
        } catch (IOException | InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

 

/**
 * AIO服务端
 *
 * @author xiangwang
 */
public class AioServer {
    public void start() throws InterruptedException, IOException {
        AsynchronousServerSocketChannel channel = AsynchronousServerSocketChannel.open();
        if (channel.isOpen()) {
            // socket接受缓冲区recbuf大小
            channel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024);
            // 端口重用,防止进程意外终止,未释放端口,重启时失败
            // 因为直接杀进程,没有显式关闭套接字来释放端口,会等待一段时间后才可以重新use这个关口
            // 解决办法就是用SO_REUSEADDR
            channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
            channel.bind(new InetSocketAddress(8080));
        } else {
            throw new RuntimeException("channel not opened!");
        }
        // 处理client连接
        channel.accept(null, new AioServerHandler(channel));
        System.out.println("server started");
        // 阻塞主进程
        for(;;) {
            TimeUnit.SECONDS.sleep(1);
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        AioServer server = new AioServer();
        server.start();
    }
}

 

/**
 * AIO服务端CompletionHandler
 *
 * @author xiangwang
 */
public class AioServerHandler implements CompletionHandler<AsynchronousSocketChannel, Void> {
    private final AsynchronousServerSocketChannel serverChannel;
    private final CharsetDecoder decoder = Charset.defaultCharset().newDecoder();
    private final BufferedReader input = new BufferedReader(new InputStreamReader(System.in));

    public AioServerHandler(AsynchronousServerSocketChannel serverChannel) {
        this.serverChannel = serverChannel;
    }
    @Override
    public void failed(Throwable exc, Void attachment) {
        // 处理下一次的client连接
        serverChannel.accept(null, this);
    }
    @Override
    public void completed(AsynchronousSocketChannel result, Void attachment) {
        // 处理下一次的client连接,类似链式调用
        serverChannel.accept(null, this);
        try {
            // 将输入内容写到buffer
            String line = input.readLine();
            result.write(ByteBuffer.wrap(line.getBytes()));
            // 在操作系统中的Java本地方法native已经把数据写到了buffer中
            // 这里只需要一个缓冲区能接收就行了
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            while (result.read(buffer).get() != -1) {
                buffer.flip();
                System.out.println("from client: " + decoder.decode(buffer).toString());
                if (buffer.hasRemaining()) {
                    buffer.compact();
                } else {
                    buffer.clear();
                }
                // 将输入内容写到buffer
                line = input.readLine();
                result.write(ByteBuffer.wrap(line.getBytes()));
            }
        } catch (InterruptedException | ExecutionException | IOException e) {
            e.printStackTrace();
        }
    }
}

 

 

执行测试后显示,不管是在客户端还是在服务端,读写完全是异步的。

 

 


 

 

感谢您的大驾光临!咨询技术、产品、运营和管理相关问题,请关注后留言。欢迎骚扰,不胜荣幸~