okio 超时机制（很好的一次多线程编程学习机会）

作者: 一个追寻者的故事 | 来源:发表于2020-08-27 15:01 被阅读0次

okio 超时机制（很好的一次多线程编程学习机会）
OkHttp之Okio源码分析(三）Okio读写流程梳理
Okio 超时机制
进程与线程
Okio 源码解析（二）：超时机制
基本的线程机制—Java编程思想
Python 3 多线程编程
基于Redis的实现分布式锁(本文提供两种方式)
读书笔记——粗读《Java并发编程的艺术》
后端架构师技术图谱(三)-并发、锁、设计模式(二)

okio 超时机制的原理（基于okio-2.7.0）

1.1 干什么用的

了解一项技术之前肯定要先知道它是解决什么问题的，也就是干什么用的。

我们先看一下实现超时机制的核心类 AsyncTimeout 的类注释，就知道超时机制解决了什么问题：

AsyncTimeout 使用了后台线程，能够在超时发生的时候准确的采取行动。使用它可以实现那些
本身不支持超时时间的情况。例如：Socket会在 write 方法执行的时候阻塞，且不支持超时时间。

总结：实现了一套超时机制，能够对于耗时、阻塞方法进行超时控制，其实是一套通用的机制，后边会演示如何使用。

把类 AsyncTimeout 的注释剩下部分补全：

子类应该事先 [timedOut]方法，当超时发生时去采取行动。这个方法会被共享的WatchDog后台线程调用。
所以不应该在[timedOut]中做耗时的操作，否则可能会影响其他超时时间的通知。

1.2 怎么使用

写一段代码，来演示下如何使用okio中的超时机制。

//定义一个普通业务类
open class SomeClass {
    @Volatile var isInterrupted = false

    //做一些耗时/阻塞的事情
    open fun doSomethingBlock() {
        println("start~")
        while (true) {
            if (isInterrupted) {  //可以人为打断，如果被打断，则抛出异常提示上层使用者
                throw InterruptedException("interrupted")
            }
        }
    }
}

// 定义超时时间实现类
class SomeClassAsyncTimeout(val obj: SomeClass) : AsyncTimeout() {
    //定义超时时间回调函数的处理逻辑。一般的情况是，如果超时了，打断方法调用
    override fun timedOut() {
        println("this operation is timed out")
        println("timedOut() 所在的线程： ${Thread.currentThread().name}")
        obj.isInterrupted = true    //打断耗时方法
    }
    
   //生成一个 SomeClass的代理对象
    fun delegate(): SomeClass {
        return object : SomeClass() {
            // 先执行代理方法
            override fun doSomethingBlock() {
                withTimeout { obj.doSomethingBlock() }
            }
        }
    }
}

fun main() {
    val someObj = SomeClassAsyncTimeout(SomeClass())
        .apply { timeout(4, TimeUnit.SECONDS) }  //设置超时时间
        .delegate()  //返回代理对象

    try{
        someObj.doSomethingBlock()    //执行代理对象的耗时方法
    }catch (e: Exception){
        e.printStackTrace()
        println("已经超时")
    }
    println("main - end")
}

我们看一下执行结果：

执行结果

可以发现，原本没有超时机制的方法，可以动态添加超时的行为。

1.3 超时怎么实现的

1.3.1 使用排序后的单链表组织超时时间

open class AsyncTimeout : Timeout() {
     //...
     /** The next node in the linked list.  */
     private var next: AsyncTimeout? = null

    //超时后的回调
     protected open fun timedOut() {}

     companion object {
          private var head: AsyncTimeout? = null
     }
     //...
}

AsyncTimeout 有一个next，其 Companion 对象中有一个 head，WatchDog后台线程处理一个由 AsyncTimeout 元素组成的一个单链表，这个单链表按照超时时间排序（链表最前边的最先超时）。对于单链表的操作使用synchronized(syncTimeout.class) 来保证操作互斥同步。

head 的 next 节点是第一个元素。第一个元素是下一个将要超时的元素。初始化情况下head = null，watchDog线程启动后，head会被创建（只是为了占位，真正数据节点是下一个元素），当watchDog线程空闲了 AsyncTimeout.IDLE_TIMEOUT_MILLIS 后会结束，同时 head = null

超时时间单链表示意图

1.3.2 后台线程WatchDog到底在干嘛

代码很短，我们把代码贴出来，一行一行分析。

private class Watchdog internal constructor() : Thread("Okio Watchdog") {
    init {
      isDaemon = true  // 设置为后台线程
    }

    override fun run() {
      while (true) {
        try {
          var timedOut: AsyncTimeout? = null
          // 访问链表结构时统一上锁
          synchronized(AsyncTimeout::class.java) {
            // 从单链表中获取下一个超时的节点。
            timedOut = awaitTimeout()

            // 当前队列中（单链表）中已经空了，直接让后台线程退出
            // 当下次有节点插入队列的时候，创建一个新的WatchDog吧
            if (timedOut === head) {
              head = null    //置空头结点 （头结点和单链表是同时存在的）
              return    // 结束线程
            }
          }

          // 如果节点存在则执行回调
          timedOut?.timedOut()
        } catch (ignored: InterruptedException) {
        }
      }
    }
  }


open class AsyncTimeout : Timeout() {

  companion object {
    // 从队列中（单链表）中获取一下一个超时的元素
    @Throws(InterruptedException::class)
    internal fun awaitTimeout(): AsyncTimeout? {
      // 获取下一个元素（上边说了，head和 WatchDog同时存在的，只有WatchDog自己会把 head 置空，所以此时head一定不为空）
      val node = head!!.next

       // 如果下一个元素为空，wait 空闲的时间（60s默认）后关闭线程，又或者因为队列前边插入了新的元素，会被唤醒。
      if (node == null) {
        val startNanos = System.nanoTime()
        (AsyncTimeout::class.java as Object).wait(IDLE_TIMEOUT_MILLIS)
        return if (head!!.next == null && System.nanoTime() - startNanos >= IDLE_TIMEOUT_NANOS) {
          head // 线程空闲等待的事件结束，线程后续会结束。
        } else {
          null // 有新的元素插入在队列的最前端了，结束本次循环
        }
      }
      
      // 获取这个这个节点的剩余超时时间
      var waitNanos = node.remainingNanos(System.nanoTime())

      // 此时队列还不存在超时的元素
      if (waitNanos > 0) {
        // 线程wait 刚才算出来的剩余的时间后在醒来。
        val waitMillis = waitNanos / 1000000L
        waitNanos -= waitMillis * 1000000L
        (AsyncTimeout::class.java as Object).wait(waitMillis, waitNanos.toInt())
        return null    //醒来后直接结束本次循环
      }

      // 这个node节点超时了，从队列中删除此结点
      head!!.next = node.next
      node.next = null
      return node
    }
  }
}

小结：WatchDog 这个后台线程，持续从超时队列中获取下一个超时的元素（这个过程是做了多线程同步的），

如果队列中没有元素了，在等待了一段时间后，线程会结束，直到又有新的节点插入队列时才会创建新的WatchDog后台线程；如果在空闲等待的这一段时间里，有元素插入到队列中，WatchDog就会被唤醒，继续 持续从超时队列中获取下一个超时的元素
如果队列中有元素，获取第一个元素（最前边的），如果发现此结点没有超时，算出还有多久超时，则再 wait 这一小段的时间后再来处理，当然这个过程中也有可能被唤醒，因为也有可能有新的节点插入了最气边；如果发现此结点已超时，则直接移除队列，然后进行 timedOut 回调。

所以 WatchDog大部分时间都是在 waiting 在状态，不消耗 cpu 资源。

1.3.3 用户的方法和 `AsyncTimeout` 进行关联

讲了这么大一堆，到底如何把用户的方法扩展成带有超时机制的呢。看上边关于扩展 SomeClass 的例子中，通过 withTimeout 就会让原来的方法具有超时性。

分析一下 withTimeout

/**
block：就是具体耗时的操作
在执行 block 之前，先执行enter，block执行之后，执行exit，如果超时发生了，会抛出一个Exception，来自 [newTimeoutException] 
   */
  inline fun <T> withTimeout(block: () -> T): T {
    var throwOnTimeout = false
    enter()    // *****
    try {
      val result = block()
      throwOnTimeout = true
      return result
    } catch (e: IOException) {  // 如果block()调用过程中抛出了Exception
      //如果已经超时，会包装成新的超时Exception，如果没有超时，直接抛出方法本身的异常
      throw if (!exit()) e else `access$newTimeoutException`(e)
    } finally {  // 即便block没有抛出异常
      val timedOut = exit()      
      //判断超时后，依然会告知用户
      if (timedOut && throwOnTimeout) throw `access$newTimeoutException`(null)
    }
  }

围绕用户耗时方法的前后，有连个方法 enter、exit
enter ：会将耗时操作对应 AsyncTimeout节点按照超时时间插入到队列的合适位置。
exit：如果此AsyncTimeout 节点还在队列中，则删除此节点（因为任务已完成，不需要在被监听）。同时会返回此 AsyncTimeout 节点是否已经超时（如果节点中已经没有此节点了，这说明此节点被WatchDog删除了，也就是超时了）。

着重看那一下入队列和出队列的方法，其它细节最好自己看源码。
入队列：

private fun scheduleTimeout(node: AsyncTimeout, timeoutNanos: Long, hasDeadline: Boolean) {
      synchronized(AsyncTimeout::class.java) {

        // 当head被创建的时候，启动WatchDog线程
        if (head == null) {
          head = AsyncTimeout()
          Watchdog().start()
        }

        // 根据用户设置的 timeoutNanos、deadlineNanoTime 转化成 timeoutAt
        val now = System.nanoTime()
        if (timeoutNanos != 0L && hasDeadline) {
          // Compute the earliest event; either timeout or deadline. Because nanoTime can wrap
          // around, minOf() is undefined for absolute values, but meaningful for relative ones.
          node.timeoutAt = now + minOf(timeoutNanos, node.deadlineNanoTime() - now)
        } else if (timeoutNanos != 0L) {
          node.timeoutAt = now + timeoutNanos
        } else if (hasDeadline) {
          node.timeoutAt = node.deadlineNanoTime()
        } else {
          throw AssertionError()
        }

        // Insert the node in sorted order.
        val remainingNanos = node.remainingNanos(now)
        var prev = head!!
        while (true) {
          if (prev.next == null || remainingNanos < prev.next!!.remainingNanos(now)) {  // 找到合适的超时时间位置落座
            //插入节点
            node.next = prev.next    
            prev.next = node
            if (prev === head) {  // 如果此时是第一个元素（有可能是第一次放，也有可能是好几轮以后了）
              // Wake up the watchdog when inserting at the front.
             // 记得这时候唤醒WatchDog
              (AsyncTimeout::class.java as Object).notify()
            }
            break
          }
          prev = prev.next!!    //一直往后找
        }
      }
    }

出队列：

 /** 返回true，如果超时已经发生 */
    private fun cancelScheduledTimeout(node: AsyncTimeout): Boolean {
      synchronized(AsyncTimeout::class.java) {
        // Remove the node from the linked list.
       // 从但链表里删除此节点
        var prev = head
        while (prev != null) {
          if (prev.next === node) {
            prev.next = node.next
            node.next = null
            return false
          }
          prev = prev.next
        }

        // The node wasn't found in the linked list: it must have timed out!
        // 此节点没有找到，说明它对应的耗时方法已经超时了
        return true
      }
    }

至此，整个处理超时的过程比较明朗了。

timedOut 这个方法会在超时后被WatchDog回调。在这个回调里，可以打断耗时方法的执行，从而让耗时方法抛出异常终止方法继续执行，告知使用者已超时；也可以不坐任何操作，等到方法执行完成以后，也会抛出异常告知使用者已经超时。这里说的打断耗时方法的执行，前提是耗时方法支持被打断，否则也无法中断程序。超时机制的设置本来是为了给okhttp用的，例如 Socket 的write 方法没有超时支持，可以通过 AsyncTimeout 监听超时之后调用 socket#close 方法继而中断 write 操作。

注意 WatchDog 是个 Daemon Thread。 Java的线程分为两种：User Thread(用户线程)、DaemonThread(守护线程)。只要当前JVM实例中尚存任何一个非守护线程没有结束，守护线程就全部工作；只有当最后一个非守护线程结束时，守护线程随着JVM一同结束工作，Daemon作用是为其他线程提供便利服务，守护线程最典型的应用就是GC(垃圾回收器)，他就是一个很称职的守护者。用户线程和守护线程两者几乎没有区别，唯一的不同之处就在于虚拟机的离开：如果用户线程已经全部退出运行了，只剩下守护线程存在了，虚拟机也就退出了。因为没有了被守护者，守护线程也就没有工作可做了，也就没有继续运行程序的必要了。

这里涉及到 wait 、 notify 、synchronized 的多线程知识，是很好的一次多线程学习机会。

从本篇文章中可以看到等待/通知的经典范式。分为两部分：等待方（消费者）和通知方（生产者）
等待方遵循原则：
1、获取对象的锁
2、如果条件不满足，那么调用对象的wait方法，被通知后仍要检查条件。
3、条件满足则执行对应的逻辑。
通知方遵循原则：
1、获得对象的锁
2、改变条件
3、通知所有等待在对象上的线程。

okio 超时机制（很好的一次多线程编程学习机会）
okio 超时机制的原理（基于okio-2.7.0） 1.1 干什么用的了解一项技术之前肯定要先知道它是解决什么...
OkHttp之Okio源码分析(三）Okio读写流程梳理
引言前面两篇文章分析了Okio的核心类和超时机制，了解了底层Okio的工作机制，今天我们自顶向下梳理Okio的I...
Okio 超时机制
Okio中有同步超时和异步超时。同步超时Timeout类 Timeout类提供了两个互补的控制机制来定义超时策略...
进程与线程
Java编程属于一种多线程的运行机制，这种多线程的编程机制在早起出现的时候绝对是一个编程语言的亮点所在，但是随着后...
Okio 源码解析（二）：超时机制
简介上一篇文章（Okio 源码解析（一）：数据读取流程）分析了 Okio 数据读取的流程，从中可以看出 Okio...
基本的线程机制—Java编程思想
基本的线程机制—Java编程思想并发编程使我们可以将程序分为多个分离的、独立运行的任务。通过使用多线程机制，这些...
Python 3 多线程编程
本文主要基于python 3.5实现多线程编程 1. 创建多线程 2. 多线程间的同步，lock机制 3. que...
基于Redis的实现分布式锁(本文提供两种方式)
关于分布式锁我们学习过并发编程中的锁机制：synchronized和Lock。在单进程的系统中，当多线程同时修改...
读书笔记——粗读《Java并发编程的艺术》
主要内容摘抄自《Java并发编程的艺术》 Java并发机制的底层实现原理在多线程并发编程中synchronize...
后端架构师技术图谱(三)-并发、锁、设计模式(二)
并发多线程《40个Java多线程问题总结》线程安全《Java并发编程——线程安全及解决机制简介》一致性、...