一丶IO模型&Java IO

Unix为程序员提供了以下5种基本的io模型：

  • blocking io： 阻塞io
  • nonblocking io： 非阻塞io
  • I/O multiplexing： io多路复用
  • signal driven I/O：信号驱动io
  • asynchronous I/O：异步io

但我们平时工作中说的最多是，
阻塞
，
非阻塞
，
同步
，
异步

1.阻塞非阻塞，同步异步

  • 阻塞
    调用是指调用结果返回之前，当前线程会被挂起。调用线程只有在得到结果之后才会返回。
    非阻塞
    调用指在不能立刻得到结果之前，该调用不会阻塞当前线程。

    image-20230405114445720

    可用把上面的阻塞队列看作是外卖柜

    put
    方法就是外卖员放外卖，如果容量不够那么一直等待其他用户拿走外卖，这是阻塞。

    offer
    方法也是外卖员放外卖，但是他发现容量不够的时候，返回false，然后采取其他行动，比如打电话喊你下来拿外卖。

  • 同步与异步关注的是
    消息通信机制
    。同步是发起调用在没有得到结果之前，该调用就不返回。异步是发起调用后，这个调用就直接返回了。

    消息队列中间件的作用之一就是异步，发送方将消息发送就立马返回了，不需要等待这个消息被消费者处理。

    同步就是你打电话问外卖员外卖到哪里了，外卖员告知你之前你不挂断电话。

    异步就是你外卖app上发消息问外卖员，发完消息你立马可用做其他的事情。

    异步情况下你怎么知道外卖到哪里了昵？


    • 通知

      外卖员通过平台回复你

    • 回调

      你给外卖员注册了一个回调事件——收到消息后，请回电告知，然后你调用结束，继续处理你的事情，但是外卖员收到消息后，会回调进行电话。

2.Unix的io模型

io操作分为两步：

  • 等待数据就绪

    例如读文件的过程中需要等待磁盘扫描所需数据，等待数据到达内核缓冲区

  • 将数据从内核空间拷贝到用户空间

    对于一次读取IO的操作，数据并不会直接拷贝到应用程序的缓冲区(用户空间)，它首先会被拷贝到操作系统内核的缓冲区(内核空间)中，然后才会从操作系统内核的缓冲区拷贝到应用程序的缓冲区。

2.1 blocking io阻塞io

img

首先是我们用户进行进行系统调用，产生中断，操作系统切换到内核态，随后是内核完成数据准备和数据从内核空间复制到用户空间，然后应用进程继续运行。

这里说的阻塞，是系统调用不会立即返回，而是需要阻塞知道数据准备完成，并拷贝到用户空间。

2.2 nonblocking io 非阻塞io

img

可看到，和阻塞io的区别在于，准备数据的这个过程，是应用程序不断进行系统调用，询问操作系统内核是否完成了数据准备，此系统调用不会阻塞直到数据准备完成，而是立马返回。

但是第二阶段，数据从内核空间复制到用户空间是阻塞的，这个过程通常是比较快速的，因为这时候已经有DMA控制器完成了数据从磁盘搬运到内存，只需要拷贝到用户态空间中即可。

2.3 I/O multiplexing io多路复用

img

可以看到IO多路复用的流程和
blocking io阻塞io
类似，甚至还会多一次系统调用。那么IO多路复用存在的意义是什么昵？

假设我们当前的进程是一个服务端程序，存在多个网络io需要处理，我们需要多个线程取处理多个网络io，并且多个线程都是阻塞在系统调用上的，这是对线程资源的浪费。

io多路复用的优点就是：可以使用一个线程监听多路io，这个线程阻塞与
select
系统调用上，当多路io存在任何一个io可读的时候，线程将被唤醒，然后进行数据的拷贝，并进行处理，从而节省线程资源。

2.4 signal driven I/O信号驱动io

img

可以看到，信号驱动的io在数据准备阶段是非阻塞的，当操作系统完成数据准备后将发送信号来通知用户进程发生了某事件,用户进程需要编写对应的信号处理函数，在信号处理函数中阻塞与内核数据拷贝，待拷贝完成后对数据进行处理。

2.5 asynchronous I/O 异步io

img

上面四种模型都会在
数据从内核空间，拷贝到用户空间
这一步发生阻塞，也就是说至少第二步是需要同步等待操作系统完成拷贝的。

异步io模型则解决了这个问题，应用程序只要通知内核要读取的套接字对象, 以及数据的接收地址, 则整个过程都是由内核独立来完成, 包括数据从内核空间向用户空间的拷贝，拷贝完成后再通过信号来通知用户进程。

2.java中的io模型

将
阻塞
，
非阻塞
，
同步
，
异步
进行组合

  • 阻塞同步io

    这就是java中的BIO

  • 非阻塞同步io

    这就是java中的NIO，java中的nio是通过io多路复用实现的

  • 非阻塞异步io

    这就是java中的AIO，java中的AIO也是通过io多路复用实现，呈现出异步的表象

二丶Java BIO

下面探讨下java中BIO实现Socket编程方面的不足

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

    ExecutorService threadPool 
            = new ThreadPoolExecutor(10,10,100, TimeUnit.SECONDS,new ArrayBlockingQueue<>(100));

    // 1 创建一个socket server监听tcp 6666661端口
    ServerSocket serverSocket = new ServerSocket(6666661);
    // 2 阻塞式接受来自客户端的连接
    while (true) {
        //这一步是阻塞的  阻塞直到有客户端连接上来
        Socket socket = serverSocket.accept();
        System.out.println(socket.getRemoteSocketAddress() + "连接到服务端");
        // 3 为了不影响后续连接进来处理，使用多线程来处理连接
        threadPool.execute(() -> process(socket));
    }
}

private static void process(Socket socket) {
    try (OutputStream out = socket.getOutputStream()) {
        byte[] buffer = new byte[1024];
        int len;
        while ((len = socket.getInputStream().read(buffer)) > 0) {
            System.out.println(socket.getRemoteSocketAddress() + "发送数据：" + new String(buffer, 0, len));
            out.write(buffer, 0, len);
        }
    } catch (Exception e) {
        e.printStackTrace();
    }
}

上面代码实现了，如果客户端请求过来讲客户端请求原封不动的写回，可以看到为了实现实现服务端支持多个客户端的连接，我们使用的线程池。

首先
Socket socket = serverSocket.accept()
,这一步会阻塞直到有客户端连接上来（这一步无所谓，甚至避免了主线程无休止的自旋）

其次
process
方法中拿到输入输出流写回的操作也是阻塞的，这一步需要使用操作系统提供的系统调用，将数据从网卡或者硬盘读入内核空间，然后从内核空间拷贝到用户空间，我们的java程序才可以进行读操作，写则反之。

由于
read,write
这两个方法是阻塞式的，它需要阻塞直到系统调用完成，我们的程序傻傻阻塞等待，因此我们使用了线程池，希望一个线程处理一个客户端请求，阻塞也只阻塞线程池中的线程。但是
process
方法中阻塞的这部分，会体现在我们线程池的线程，也就是说，线程池中存在一些线程阻塞于
read
,
write
函数。

这种模型的优点：

  • 简单直接，可以让开发人员专注于编写process的业务逻辑
  • 不用过多考虑系统的过载、限流等问题。线程池本身就是一个天然的漏斗，可以缓冲一些系统处理不了的连接或请求。
  • 使用多线程利用多核心cpu的能力，当线程阻塞的时候，cpu可以切换时间片给其他线程

这种模型的缺点：

  • 非常依赖于线程，线程是宝贵的资源，虽然使用线程池进行了复用，当前当大量请求到来的时候，我们无法无限制的开辟线程。众多的线程被挂起，被唤醒还会导致上下文切换频繁，cpu利用率降低
  • 线程本身占用较大内存，过多的线程导致jvm内存岌岌可危

那么怎么解决上述的问题昵，能不能解放线程不让他们阻塞在read和write中，能读那就读，不能读那就继续处理其他socket？

三丶Java NIO

image-20230405164149306

回顾这张图，我们上面说的
解放线程不让他们阻塞在read和write中，能读那就读，不能读那就继续处理其他socket
,不正是上面非阻塞的方式，希望系统调用可以立即返回，而不是阻塞。

Java中的nio基于io多路复用实现了同步非阻塞的处理方式

public static void main(String[] args) throws IOException, InterruptedException {
    // 1 创建selector用来侦听多路IO消息 '文件描述符'
    // selector 担任了重要的通知角色，可以将任意IO注册到selector上，通过非阻塞轮巡selector来得知哪些路IO有消息了
    // 底层是epoll（linux下）
    // 后续会把server端注册上来，有服务端被客户端连接上来的IO消息
    // 也会把每个客户端连接注册上来，有客户端发送过来的数据
    Selector selector = Selector.open();
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

    // 2 把server端注册上去
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    serverSocketChannel.socket().bind(new InetSocketAddress("127.0.0.1", 6666661));
    //配置为非阻塞
    serverSocketChannel.configureBlocking(false);
    //关心accept事件，
    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    while (true) {
        // 3 这一步是阻塞的，基于io多路复用中的select poll，epoll
        // 这里可以设置等待事件
        if (selector.select() == 0) {
            continue;
        }

        // 4 如果有至少一路IO有消息，那么set不为空
        Set<SelectionKey> selectionKeys = selector.selectedKeys();
        for (SelectionKey key : selectionKeys) {
            if (key.isAcceptable()) {
                System.out.println("客户端连接");
                // 因为我们只注册了serverSocketChannel这一个可以accept的所以这里用强转即可
                SocketChannel socketChannel = ((ServerSocketChannel) key.channel()).accept();
                socketChannel.configureBlocking(false);
                // 5 当第一次客户端连接时，就将这个连接也作为channel注册上，他是可读型的
                //当前只是有客户端连接上来了，但是并不代表可读，还需要DMA将网卡数据搬运到内存
                socketChannel.register(selector, SelectionKey.OP_READ);
            } else if (key.isReadable()) {
                // 6 因为步骤5把客户端连接也注册上来了，并且是可读上面的数据的，如果该channel被选出来说明有客户端数据来了
                SocketChannel socketChannel = (SocketChannel) key.channel();
                // 7 必须借助ByteBuffer接受和发送数据
                byteBuffer.clear();
                if (socketChannel.read(byteBuffer) <= 0) {
                    continue;
                }
                byteBuffer.flip();
                byte[] b = new byte[byteBuffer.limit()];
                byteBuffer.get(b);
                System.out.println(key + " 数据来了： " + new String(b));
                byteBuffer.clear();
                byteBuffer.put(b);
                byteBuffer.flip();
                socketChannel.write(byteBuffer);
            }
        }
        // 8 非常重要一定要清理掉每个channel的key，来表示已经处理过了，不然下一次还会被select
        selectionKeys.clear();
    }
}

select是阻塞的，无论是通过操作系统的通知（epoll）还是不停的轮询(select，poll)，这个函数是阻塞的，它还支持超时阻塞模式。这是一个线程监听多路io的体现，只要有一个事件就绪那么select就会返回。

socketChannel.configureBlocking(false)
将 socketChannel设置为非阻塞其读写操作都是非阻塞的，也就说如果无法读，那么read函数返回-1，将会让当前线程去遍历其他就绪的事件，而不是傻傻等待，这是
非阻塞io的体现
。

四丶Java AIO

 public static void main(String[] args) throws IOException, InterruptedException {
        AsynchronousServerSocketChannel serverChannel =
                AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(6666661));
        System.out.println(Thread.currentThread() + "开始监听6666661端口");
        serverChannel.accept(null, new CompletionHandler<>() {
            @SneakyThrows
            @Override
            public void completed(AsynchronousSocketChannel channel, Object attachment) {
                // 递归注册accept
                serverChannel.accept(attachment, this);
                System.out.println(Thread.currentThread() + "有客户端连接上来了" + channel.getRemoteAddress());
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                channel.read(buffer, null, new CompletionHandler<Integer, ByteBuffer>() {
                    @SneakyThrows
                    @Override
                    public void completed(Integer len, ByteBuffer attachment) {
                        // 递归注册read
                        channel.read(buffer, null, this);
                        buffer.flip();
                        System.out.println(channel.getRemoteAddress() + "：" + new String(buffer.array(), 0, len));
                        buffer.clear();
                        channel.write(ByteBuffer.wrap("HelloClient".getBytes()));
                    }

                    @Override
                    public void failed(Throwable exc, ByteBuffer attachment) {

                    }
                });
            }

            @Override
            public void failed(Throwable exc, Object attachment) {
            }
        });
        Thread.sleep(Integer.MAX_VALUE);
    }

在
AIO
中，所有创建的通道都会直接在
OS
上注册监听，当出现
IO
请求时，会先由操作系统接收、准备、拷贝好数据，然后再通知监听对应通道的程序处理数据。

客户端的连接到来后同样会先注册到选择器上，但客户端的
I/O
请求会先交由
OS
处理，当内核将数据拷贝完成后才会分配一条线程处理。这一点不同于BIO和NIO，NIO和BIO在内核拷贝数据到用户态的这一步任然是阻塞的。

标签: none

添加新评论