Kotlin 协程 future 等待超时（不可取消）

鉴于我们有一个CompletableFuture f，在 kotlin 可挂起范围内我们可以调用f.await()我们将暂停直到完成。

我在使用签名实现类似功能时遇到问题f.await(t)必须暂停最大t毫秒或如果 future 在该持续时间内完成则更快返回（以先发生者为准）。

这是我尝试过的。

/**
 * Suspend current method until future is done or specified duration expires,
 * whichever happens first without cancelling the future.
 * Returns true if its done, false otherwise.
 */
suspend fun <T> ListenableFuture<T>.await(duration: Long): Boolean {
   val future = this
   try {
      withTimeout(duration) {
         withContext(NonCancellable) { // this does not help either
            future.await() // i do not expect the future itself to be cancelled
         }
      }
   } catch (t: TimeoutCancellationException) {
      // we expected this
   } catch (e: Throwable) {
      e.printStackTrace()
   }

   return future.isDone

}

fun main(args: Array<String>) = runBlocking<Unit> {
   val future = GlobalScope.future {
      try {
         repeat(5) {
            println("computing")
            delay(500)
         }
         println("complete")
      } finally {
         withContext(NonCancellable) {
            println("cancelling")
            delay(500)
            println("cancelled")
         }
      }
   }

   for (i in 0..10) {
      if (future.await(2000)) {
         println("checking : done")
      } else {
         println("checking : not done")
      }
   }
}

我的工作也需要类似的功能。但也许这个问题的解决方案也会帮助我......

其输出是

computing
computing
computing
computing
checking : done
checking : done
checking : done
checking : done
cancelling
checking : done
checking : done
checking : done
checking : done
checking : done
checking : done
checking : done

我写了一些测试代码：

fun main(args: Array<String>) = runBlocking {
    val future = calculateAsync()
    val result = future.await(2000)
    println("result=$result")
}

suspend fun <T> CompletableFuture<T>.await(duration: Long): T? {
    val future = this
    var result: T? = null
    try {
        withTimeout(duration) {
            result = future.await()
        }
    } catch (t: TimeoutCancellationException) {
        println("timeout exception")
    } catch (e: Throwable) {
        e.printStackTrace()
    }

    return result
}

@Throws(InterruptedException::class)
fun calculateAsync(): CompletableFuture<String> {
    val completableFuture = CompletableFuture<String>()

    Executors.newCachedThreadPool().submit {
        Thread.sleep(3000)
        println("after sleep")
        completableFuture.complete("Completed")
    }

    return completableFuture
}

运行此代码后，我们将得到输出：

timeout exception
result=null
after sleep

我们看到我们的扩展函数await回报null因为我们将超时设置为 2000 毫秒但是CompletableFuture3000 毫秒后完成。在这种情况下CompletableFuture被取消（其isCancelled财产回报true），但是我们运行的线程calculateAsync函数继续执行（我们在日志中看到它after sleep).

如果我们将超时时间设置为 4000 毫秒future.await(4000) in the main函数，我们将看到下一个输出：

after sleep
result=Completed

现在我们有了一些结果，因为CompletableFuture执行速度快于 4000 毫秒。