zoukankan      html  css  js  c++  java

  • Mina源码阅读笔记（六）—Mina异步IO的实现IoFuture

    IoFuture是和IoSession紧密相连的一个类，在官网上并没有对它的描述，因为它一般不会显示的拿出来用，权当是一个工具类被session所使用。当然在作用上，这个系列可并不简单，我们先看源码的注释对它的描述：

    IoFuture represents the completion of an asynchronous I/O operation on an IoSession.

    这个类是提供异步操作的一个工具，所以在读源码之前，必须对异步IO操作有所了解，然后我们才可以顺着这条路往下走。关于异步IO的介绍可以看：《同步、异步、阻塞、非阻塞》

    IoFuture通过IoFutureListener被IoSession绑定，它的实现都放在org.apache.mina.core.future下。在IoFuture的实现中，分别提供了读、写、连接、关闭的future，通过这四个future来实现异步操作。异步操作很重要的一部分就是对线程的控制，所以在IoFuture这个接口中，我们能很清楚的理清这几个方法：await、join。当然还有notify，但是notify没有必要写在接口中，它可以在程序里直接使用。

    这个系列的类设计的很规整，从上图的结构就能看出来，图中省略了write和connection的图，它们分别和read与close一致。由于这个类的操作没有那么复杂，继承关系也没有那么多层，所以这里面都没有用到Abstract的类来做具体实现。

    下面我们来看这里面最核心的一个类DefaultIoFuture。这个类实现IoFuture接口，主要实现了对await和join的操作，以及处理死锁的操作。我们先看这个类关联到的成员变量，都比较简单：

    01 /** A number of seconds to wait between two deadlock controls ( 5 seconds ) */
    02     private static final long DEAD_LOCK_CHECK_INTERVAL = 5000L;
    03  
    04     /** The associated session */
    05     private final IoSession session;
    06  
    07     /** A lock used by the wait() method */
    08     private final Object lock;
    09  
    10     private IoFutureListener<?> firstListener;
    11  
    12     private List<IoFutureListener<?>> otherListeners;
    13  
    14     private Object result;
    15  
    16     private boolean ready;
    17  
    18     private int waiters;

    在我看来，读源码的目的，一是为了理清框架的设计逻辑，理清结构，学习这些关联关系；二是为了学习处理细节，比如死锁的处理、线程安全的处理。在这个类中，我们将看到mina作者是如何处理死锁的问题的。

    我们先看await操作，await主要是为了等待异步操作的完成，然后通知相关的listener。我们先看一个简单的await操作和验证死锁的操作：

    01 public IoFuture await() throws InterruptedException {
    02         synchronized (lock) {
    03             while (!ready) {
    04                 waiters++;
    05                 try {
    06                     lock.wait(DEAD_LOCK_CHECK_INTERVAL);
    07                 } finally {
    08                     waiters--;
    09                     if (!ready) {
    10                         checkDeadLock();
    11                     }
    12                 }
    13             }
    14         }
    15         return this;
    16 }

    我们应该要注意下在await方法中的wait操作，这里讲些题外话，面试中常问的wait和sleep的区别。wait的操作其实很规范，必须写在synchronized块内，必须由其他线程来notify，同时wait释放锁，不占资源。而sleep占着cup的资源区睡眠，时间没到不能被唤醒，只能通过中断来打断。在这个await方法中，wait了check dead lock的时间，并且设置了计数器waiters。这个waiters在setValue方法中被运用到，在setValue中：

    01 public void setValue(Object newValue) {
    02         synchronized (lock) {
    03             // Allow only once.
    04             if (ready) {
    05                 return;
    06             }
    07  
    08             result = newValue;
    09             ready = true;
    10             if (waiters > 0) {
    11                 lock.notifyAll();
    12             }
    13         }
    14  
    15         notifyListeners();
    16     }

    异步操作是没有一个固定的顺序，谁先做好谁就返回，所以一旦有异步任务完成了操作，就会notify所有的等待，让接下来先抢到的线程再执行。在DefaultIoFuture这个类中，我觉得最重要的到不是连接或者读写，而是上面提到的setValue和getValue，因为在后续的继承关系中，会不断的用到这两个方法。不仅在后续的继承关系中，这两个方法真正在传递值得操作是发生在IoService中，不要忘了虽然session很重要，但真正起连接作用的还是service。

    然后我们再看下上面提到的check dead lock的方法，在抢占中只有读、写和连接会产生死锁的情况：

    01 private void checkDeadLock() {
    02         if (!(this instanceof CloseFuture || this instanceof WriteFuture || this instanceof ReadFuture || this instanceof ConnectFuture)) {
    03             return;
    04         }
    05         StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
    06  
    07         // Simple and quick check.
    08         for (StackTraceElement s : stackTrace) {
    09             if (AbstractPollingIoProcessor.class.getName().equals(s.getClassName())) {
    10                 IllegalStateException e = new IllegalStateException("t");
    11                 e.getStackTrace();
    12                 throw new IllegalStateException("DEAD LOCK: " + IoFuture.class.getSimpleName()
    13                         + ".await() was invoked from an I/O processor thread.  " + "Please use "
    14                         + IoFutureListener.class.getSimpleName() + " or configure a proper thread model alternatively.");
    15             }
    16         }
    17  
    18         // And then more precisely.
    19         for (StackTraceElement s : stackTrace) {
    20             try {
    21                 Class<?> cls = DefaultIoFuture.class.getClassLoader().loadClass(s.getClassName());
    22                 if (IoProcessor.class.isAssignableFrom(cls)) {
    23                     throw new IllegalStateException("DEAD LOCK: " + IoFuture.class.getSimpleName()
    24                             + ".await() was invoked from an I/O processor thread.  " + "Please use "
    25                             + IoFutureListener.class.getSimpleName()
    26                             + " or configure a proper thread model alternatively.");
    27                 }
    28             } catch (Exception cnfe) {
    29                 // Ignore
    30             }
    31         }
    32     }

    在追踪堆栈信息时，这里采用了两种check方式，简单和精确。在简单检测中，只是对比了类名，也就是对当前类有效，是一个一对一的比较。而在精确的检测中，采用isAssignableFrom方法来分别和其父类和本类进行比较。如果有死锁，就抛异常。另外join方法被废弃，由awaitUninterruptibly代替，虽然叫join，其实还是一种wait操作，等到一定时间将flag转变一下。

    下面我们看ReadFuture接口，这个接口直接继承IoFuture接口，并添加了相关的写操作。接口由DefaultReadFuture实现。在使用中，ReadFuture在IoSession中read方法中被使用，也可以说，在session层，直接读写的是future，我们再看下AbstractIoSession中的read代码：

    01 public final ReadFuture read() {
    02         if (!getConfig().isUseReadOperation()) {
    03             throw new IllegalStateException("useReadOperation is not enabled.");
    04         }
    05  
    06         Queue<ReadFuture> readyReadFutures = getReadyReadFutures();
    07         ReadFuture future;
    08         synchronized (readyReadFutures) {
    09             future = readyReadFutures.poll();
    10             if (future != null) {
    11                 if (future.isClosed()) {
    12                     // Let other readers get notified.
    13                     readyReadFutures.offer(future);
    14                 }
    15             } else {
    16                 future = new DefaultReadFuture(this);
    17                 getWaitingReadFutures().offer(future);
    18             }
    19         }
    20  
    21         return future;
    22     }

    每次都是从队列中拿出一个future，同理，每次写也是往队列里写入一个future。在DefaultReadFuture中的方法都比较简单，这里就不贴出来了。另外WriteFuture和DefaultWriteFuture和read类似，也不再赘述。

    最后我们看看ConnectFuture，我们常常写这么一段话来拿到session：

    1 ConnectFuture future = connector.connect(new InetSocketAddress(
    2                     HOST, PORT));// 创建连接
    3             future.awaitUninterruptibly();// 等待连接创建完成
    4             session = future.getSession();// 获得session

    提一点，在多态的使用上，ConnectFuture完全可以换成IoFuture，这对后面的代码没有一点儿影响，getSession本身就是继承自IoFuture的。ConnectFuture接口由DefaultConnectFuture来具体实现，由于继承了DefaultIoFuture，所以这里面用到最多的就是DefaultIoFuture中的setValue和getValue方法，上面我们也特别强调了这两个方法的重要性，通过对result（setValue）的传递实现了对session、exception等状态的传递。

    稍微总结一下future对异步的贡献，官方对future的描述就是处理了异步操作，从源码中我们很明显的可以看到future是通过await和notify来控制操作的连续性，通过死锁检测来做wait时的保障，上层（session）通过队列来缓冲各种任务，然后通过竞争，谁抢到了线程，谁就执行。Future不难，组织结构也很清楚，我觉得看这节的源代码最主要的还是要做好两点，第一是搞懂什么是异步，第二是要明白future为异步贡献了什么。

    下一篇就要讲mina中最庞大的filter chain了，这是mina中代码最多，也是最具特色的一部分。
  • 相关阅读:
    gtest（C++单元测试框架）
    tinyXML入门
    笔记 解决vue3动态绑定本地图片失效问题
    面试技巧
    vuex 状态管理
    插槽的使用
    Vue-router 路由
    Vue组件
    (转)JS 常用 DOM
    9-26
  • 原文地址：https://www.cnblogs.com/wuyida/p/6300941.html
Copyright © 2011-2022 走看看