Java基础-IO

作者: 16325 | 来源:发表于2020-02-17 10:49 被阅读0次

概念

IO指的是输入输出，主要是指文件的IO和网络的IO。
对于java程序来说，分成BIO，NIO，AIO。
各种IO的操作流程：
1.向操作系统请求数据。
2.操作系统将外部数据复制到内核缓冲区。
3.操作系统将缓冲区数据复制到进程缓冲区。
4.进程获取数据，完成IO操作。

BIO

BIO是最早的阻塞IO，是同步阻塞IO，对应操作系统调用socket的函数，这些函数都是阻塞的，所以在BIO中，读取文件，等待网络连接，等到socket数据，这些操作统统都是阻塞的。

NIO

因为BIO的阻塞特性，所以每次网络操作，包括等待连接，获取数据，发送数据，都是阻塞的，给变成带来很大麻烦，只能是针对每一个网络客户端创建新的线程来处理，带来了严重的系统负担。
NIO是基于操作的系统的select函数，linux的poll函数进行调用，针对于IO操作流程的第二点和第三点，这两部操作对于java进程来说，都是非阻塞的。所以可以实现为：向系统注册了感兴趣的事件后，只要检查这些事件是否完成，即可得到所需的数据。
这时候可能读者要问：在我们写NIO的时候，也需要while（true）啊，如下代码：

            Selector selector = Selector.open();
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.socket().bind(new InetSocketAddress(8090));
            serverSocketChannel.configureBlocking(false);
            //向通道注册选择器，并且注册接受事件
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            while (true) {
                //获取已经准备好的通道数量
                int readyChannels = selector.selectNow();
                //如果没准备好，重试
                if (readyChannels == 0) continue;
                //获取准备好的通道中的事件集合
                Set selectedKeys = selector.selectedKeys();
                Iterator keyIterator = selectedKeys.iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey key = (SelectionKey) keyIterator.next();
                    if (key.isAcceptable()) {
                         这里注册的是accept事件，
                    } else if (key.isConnectable()) {
                    } else if (key.isReadable()) {
                    } else if (key.isWritable()) {
                    }
                    //注意每次迭代末尾的keyIterator.remove()调用。
                    //Selector不会自己从已选择键集中移除SelectionKey实例。必须在处理完通道时自己移除。
                    //下次该通道变成就绪时，Selector会再次将其放入已选择键集中
                    keyIterator.remove();
                }
            }

这里第一个while(true)难道不会把进程阻塞起来？
其实这里的阻塞和操作系统调用时候的阻塞并不是同一个含义，NIO的非阻塞指的是客户端链接，客户端数据的读写这些操作是非阻塞的，也就是说我们的java进程不需要在等待客户端链接的时候阻塞，而是有客户端链接之后，我们的进程就会可以拿到这个链接，注意这里是可以拿到，而并不是已经拿到，并没有回调的功能，如果注册了回调，客户端链接上来我们的回调函数执行，这就是AIO的范畴了。
所以，我们可以拿到链接，可以拿到客户端发来的数据，可以写数据。但是这只是可以，具体的拿这个操作，还是需要自己完成的。
这里可以想象一个极端的例子，客户端链接成功能后，五分钟内肯定会发来一条消息，那么我们真的可以不要代码中的while(true),而是等五分钟后，直接从selector中获取数据。

如果像想使用NIO一样使用BIO呢？等待连接是阻塞的，这里没办法了。链接上来之后保存这些链接，每个链接一个线程，然后阻塞的read，哪个线程read完成了，告诉另一个控制线程说我读完了。然后控制线程while（true）轮询，拿到各个客户端的数据。这里的每个客户端线程的阻塞read，read之后的挂载数据，其实就是NIO中操作系统代替我们完成的工作。

AIO

根据NIO的while(true), 如果不需要我们自己拿已完成的事件和数据，而是提供回调函数，等客户端链接完成或者数据收到，自动执行回调函数，这就是AIO了。这时候就真的是异步操作了。

Reactor模型

经过之前的基础概念的说明，可以看出NIO使用一个线程就可以完成所有感兴趣事件的注册和轮询了，那么NIO该如果设计和优化我们的服务端代码呢？
自POSA2 中的关于Reactor Pattern 介绍中，我们了解了Reactor 的处理方式:

1.同步的等待多个事件源到达（采用select()实现）
2.将事件多路分解以及分配相应的事件服务进行处理，这个分派采用server集中处理（dispatch）
3.分解的事件以及对应的事件服务应用从分派服务中分离出去（handler）

为何要是用reactor模型？
先来看一段NIO的代码：

public NIOServer(int port) throws Exception {
      selector = Selector.open();
      serverSocket = ServerSocketChannel.open();
      serverSocket.socket().bind(new InetSocketAddress(port));
      serverSocket.configureBlocking(false);
      serverSocket.register(selector, SelectionKey.OP_ACCEPT);
  }

  @Override
  public void run() {
      while (!Thread.interrupted()) {
          try {
              //阻塞等待事件
              selector.select();
              // 事件列表
              Set selected = selector.selectedKeys();
              Iterator it = selected.iterator();
              while (it.hasNext()) {
                  it.remove();
                  //分发事件
                  dispatch((SelectionKey) (it.next()));
              }
          } catch (Exception e) {

          }
      }
  }

  private void dispatch(SelectionKey key) throws Exception {
      if (key.isAcceptable()) {
          register(key);//新链接建立，注册
      } else if (key.isReadable()) {
          read(key);//读事件处理
      } else if (key.isWritable()) {
          wirete(key);//写事件处理
      }
  }

  private void register(SelectionKey key) throws Exception {
      ServerSocketChannel server = (ServerSocketChannel) key
              .channel();
      // 获得和客户端连接的通道
      SocketChannel channel = server.accept();
      channel.configureBlocking(false);
      //客户端通道注册到selector 上
      channel.register(this.selector, SelectionKey.OP_READ);
  }

我们可以看到上述的NIO例子已经差不多拥有reactor的影子了:

1.基于事件驱动-> selector（支持对多个socketChannel的监听）
2.统一的事件分派中心-> dispatch
3.事件处理服务-> read & write

事实上NIO已经解决了上述BIO暴露的一些问题了，服务器的并发客户端有了量的提升，不再受限于一个客户端一个线程来处理，而是一个线程可以维护多个客户端（selector 支持对多个socketChannel 监听）。但这依然不是一个完善的Reactor Pattern ,首先Reactor 是一种设计模式，好的模式应该是支持更好的扩展性，显然以上的并不支持，另外好的Reactor Pattern 必须有以下特点：

1.更少的资源利用，通常不需要一个客户端一个线程
2.更少的开销，更少的上下文切换以及locking
3.能够跟踪服务器状态
4.能够管理handler 对event的绑定

那么好的Reactor Pattern应该是怎样的？
首先我们基于Reactor Pattern 处理模式中，定义以下三种角色:

Reactor将I/O事件分派给对应的Handler

Acceptor处理客户端新连接，并分派请求到处理器链中
Handlers执行非阻塞读/写任务

单Reactor单线程模型

image.png

/**
    * 等待事件到来，分发事件处理
    */
  class Reactor implements Runnable {

      private Reactor() throws Exception {

          SelectionKey sk =
                  serverSocket.register(selector,
                          SelectionKey.OP_ACCEPT);
          // attach Acceptor 处理新连接
          sk.attach(new Acceptor());
      }

      public void run() {
          try {
              while (!Thread.interrupted()) {
                  selector.select();
                  Set selected = selector.selectedKeys();
                  Iterator it = selected.iterator();
                  while (it.hasNext()) {
                      it.remove();
                      //分发事件处理
                      dispatch((SelectionKey) (it.next()));
                  }
              }
          } catch (IOException ex) {
              //do something
          }
      }

      void dispatch(SelectionKey k) {
          // 若是连接事件获取是acceptor
          // 若是IO读写事件获取是handler
          Runnable runnable = (Runnable) (k.attachment());
          if (runnable != null) {
              runnable.run();
          }
      }

  }
  /**
    * 连接事件就绪,处理连接事件
    */
  class Acceptor implements Runnable {
      @Override
      public void run() {
          try {
              SocketChannel c = serverSocket.accept();
              if (c != null) {// 注册读写
                  new Handler(c, selector);
              }
          } catch (Exception e) {

          }
      }
  }
  /**
    * 处理读写业务逻辑
    */
  class Handler implements Runnable {
      public static final int READING = 0, WRITING = 1;
      int state;
      final SocketChannel socket;
      final SelectionKey sk;

      public Handler(SocketChannel socket, Selector sl) throws Exception {
          this.state = READING;
          this.socket = socket;
          sk = socket.register(selector, SelectionKey.OP_READ);
          sk.attach(this);
          socket.configureBlocking(false);
      }

      @Override
      public void run() {
          if (state == READING) {
              read();
          } else if (state == WRITING) {
              write();
          }
      }

      private void read() {
          process();
          //下一步处理写事件
          sk.interestOps(SelectionKey.OP_WRITE);
          this.state = WRITING;
      }

      private void write() {
          process();
          //下一步处理读事件
          sk.interestOps(SelectionKey.OP_READ);
          this.state = READING;
      }

      /**
        * task 业务处理
        */
      public void process() {
          //do something
      }
  }

这是最基本的单Reactor单线程模型。其中Reactor线程，负责多路分离套接字，有新连接到来触发connect 事件之后，交由Acceptor进行处理，有IO读写事件之后交给hanlder 处理。Acceptor主要任务就是构建handler ，在获取到和client相关的SocketChannel之后，绑定到相应的hanlder上，对应的SocketChannel有读写事件之后，基于reactor 分发,hanlder就可以处理了（所有的IO事件都绑定到selector上，有Reactor分发）。该模型适用于处理器链中业务处理组件能快速完成的场景。不过，这种单线程模型不能充分利用多核资源，所以实际使用的不多。

单Reactor多线程模型

image.png

/**
    * 多线程处理读写业务逻辑
    */
  class MultiThreadHandler implements Runnable {
      public static final int READING = 0, WRITING = 1;
      int state;
      final SocketChannel socket;
      final SelectionKey sk;

      //多线程处理业务逻辑
      ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());


      public MultiThreadHandler(SocketChannel socket, Selector sl) throws Exception {
          this.state = READING;
          this.socket = socket;
          sk = socket.register(selector, SelectionKey.OP_READ);
          sk.attach(this);
          socket.configureBlocking(false);
      }

      @Override
      public void run() {
          if (state == READING) {
              read();
          } else if (state == WRITING) {
              write();
          }
      }

      private void read() {
          //任务异步处理
          executorService.submit(() -> process());

          //下一步处理写事件
          sk.interestOps(SelectionKey.OP_WRITE);
          this.state = WRITING;
      }

      private void write() {
          //任务异步处理
          executorService.submit(() -> process());

          //下一步处理读事件
          sk.interestOps(SelectionKey.OP_READ);
          this.state = READING;
      }

      /**
        * task 业务处理
        */
      public void process() {
          //do IO ,task,queue something
      }
  }

相对于第一种单线程的模式来说，在处理业务逻辑，也就是获取到IO的读写事件之后，交由线程池来处理，这样可以减小主reactor的性能开销，从而更专注的做事件分发工作了，从而提升整个应用的吞吐。

多Reactor多线程模型

image.png

/**
    * 多work 连接事件Acceptor,处理连接事件
    */
  class MultiWorkThreadAcceptor implements Runnable {

      // cpu线程数相同多work线程
      int workCount =Runtime.getRuntime().availableProcessors();
      SubReactor[] workThreadHandlers = new SubReactor[workCount];
      volatile int nextHandler = 0;

      public MultiWorkThreadAcceptor() {
          this.init();
      }

      public void init() {
          nextHandler = 0;
          for (int i = 0; i < workThreadHandlers.length; i++) {
              try {
                  workThreadHandlers[i] = new SubReactor();
              } catch (Exception e) {
              }

          }
      }

      @Override
      public void run() {
          try {
              SocketChannel c = serverSocket.accept();
              if (c != null) {// 注册读写
                  synchronized (c) {
                      // 顺序获取SubReactor，然后注册channel 
                      SubReactor work = workThreadHandlers[nextHandler];
                      work.registerChannel(c);
                      nextHandler++;
                      if (nextHandler >= workThreadHandlers.length) {
                          nextHandler = 0;
                      }
                  }
              }
          } catch (Exception e) {
          }
      }
  }
  /**
    * 多work线程处理读写业务逻辑
    */
  class SubReactor implements Runnable {
      final Selector mySelector;

      //多线程处理业务逻辑
      int workCount =Runtime.getRuntime().availableProcessors();
      ExecutorService executorService = Executors.newFixedThreadPool(workCount);


      public SubReactor() throws Exception {
          // 每个SubReactor 一个selector 
          this.mySelector = SelectorProvider.provider().openSelector();
      }

      /**
        * 注册chanel
        *
        * @param sc
        * @throws Exception
        */
      public void registerChannel(SocketChannel sc) throws Exception {
          sc.register(mySelector, SelectionKey.OP_READ | SelectionKey.OP_CONNECT);
      }

      @Override
      public void run() {
          while (true) {
              try {
              //每个SubReactor 自己做事件分派处理读写事件
                  selector.select();
                  Set<SelectionKey> keys = selector.selectedKeys();
                  Iterator<SelectionKey> iterator = keys.iterator();
                  while (iterator.hasNext()) {
                      SelectionKey key = iterator.next();
                      iterator.remove();
                      if (key.isReadable()) {
                          read();
                      } else if (key.isWritable()) {
                          write();
                      }
                  }

              } catch (Exception e) {

              }
          }
      }

      private void read() {
          //任务异步处理
          executorService.submit(() -> process());
      }

      private void write() {
          //任务异步处理
          executorService.submit(() -> process());
      }

      /**
        * task 业务处理
        */
      public void process() {
          //do IO ,task,queue something
      }
  }

第三种模型比起第二种模型，是将Reactor分成两部分:

mainReactor负责监听server socket，用来处理新连接的建立，将建立的socketChannel指定注册给subReactor。
subReactor维护自己的selector, 基于mainReactor 注册的socketChannel多路分离IO读写事件，读写网络数据，对业务处理的功能，另其扔给worker线程池来完成。

第三种模型中，我们可以看到，mainReactor 主要是用来处理网络IO 连接建立操作，通常一个线程就可以处理，而subReactor主要做和建立起来的socket做数据交互和事件业务处理操作，它的个数上一般是和CPU个数等同，每个subReactor一个县城来处理。此种模型中，每个模块的工作更加专一，耦合度更低，性能和稳定性也大量的提升，支持的可并发客户端数量可达到上百万级别。关于此种模型的应用，目前有很多优秀的矿建已经在应用了，比如mina 和netty 等。上述中去掉线程池的第三种形式的变种，也是Netty NIO的默认模式。下一节我们将着重讲解netty的架构模式。

Java基础-IO

概念

BIO

NIO

AIO

Reactor模型

单Reactor单线程模型

单Reactor多线程模型

多Reactor多线程模型

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读