Kotlin 协程未来等待超时（无取消）

Question

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

我在实现带有签名 f.await(t) 的类似函数时遇到问题，该函数必须暂停最多 t 毫秒，或者如果未来在该持续时间内完成（以先发生者为准），则更快返回。

这是我尝试过的。

/**
 * 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

Answer 1

我已经写了一些测试代码：

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 毫秒，但 CompletableFuture 在 3000 毫秒后完成。在这种情况下CompletableFuture 被取消（它的isCancelled 属性返回true），但是我们在calculateAsync 函数中运行的线程继续执行（我们在日志中看到它after sleep）。

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

after sleep
result=Completed

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

Answer 2

这是我想出的，我认为这不是一个好的解决方案，因为我很可能会为相当原始的任务创建大量垃圾。

suspend fun <T> CompletableFuture<T>.await(duration: Millis): Boolean {
   val timeout = CompletableFuture<Unit>()

   GlobalScope.launch {
      delay(duration)
      timeout.complete(Unit)
   }

   val anyOfTwo = CompletableFuture.anyOf(this, timeout)
   anyOfTwo.await()
   return this.isDone
}


fun main() = runBlocking {
   val future = CompletableFuture<String>()

   GlobalScope.launch {
      delay(2000)
      println("setting the result (future now ${future.isDone})")
      future.complete("something")
   }

   while (future.isNotDone()) {
      println("waiting for the future to complete for the next 500ms")
      val isDone = future.await(500)

      if (isDone) {
         println("future is done")
         break
      } else {

         println("future not done")
      }
   }

   Unit
}

这将给出

的输出

waiting for the future to complete for the next 500ms
future not done
waiting for the future to complete for the next 500ms
future not done
waiting for the future to complete for the next 500ms
future not done
waiting for the future to complete for the next 500ms
setting the result (future now false)
future is done

这正是我们想要的……