Scala异步请求的并发超时处理指南

import scala.concurrent.{Future, Promise}
import scala.concurrent.duration.Duration
import scala.concurrent.ExecutionContext.Implicits.global // 导入默认的ExecutionContext
import com.twitter.util.{Scheduler => TwitterScheduler} // 假设使用一个调度器，如Twitter Util的Scheduler

// 定义一个通用的Scheduler接口，或者直接使用java.util.concurrent.ScheduledExecutorService
// 这里为了演示，我们假设存在一个类似Twitter Util的Scheduler
object Scheduler {
  def after(d: Duration)(block: => Unit): Unit = {
    // 实际实现会使用一个定时任务服务，例如ScheduledExecutorService
    // 示例：在一个延迟后执行block
    val executor = java.util.concurrent.Executors.newSingleThreadScheduledExecutor()
    executor.schedule(new Runnable {
      override def run(): Unit = block
    }, d.toMillis, java.util.concurrent.TimeUnit.MILLISECONDS)
    executor.shutdown()
  }
}

/**
 * 创建一个Future，它将在给定持续时间后成功完成并携带None。
 * @param d 超时持续时间。
 * @return 一个Future[Option[Nothing]]，在超时后完成。
 */
def timeout(d: Duration): Future[Option[Nothing]] = {
  val p = Promise[Option[Nothing]]()
  Scheduler.after(d) {
    p success None
  }
  p.future
}

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global // 导入ExecutionContext

/**
 * 组合两个Future，返回其中第一个完成的结果。
 * 如果f1先完成，返回Some(f1的结果)；如果f2（超时Future）先完成，返回f2的结果（通常是None）。
 * @param f1 原始请求的Future。
 * @param f2 超时Future，通常是timeout函数返回的Future[Option[Nothing]]。
 * @tparam T f1的类型。
 * @return 一个Future[Option[T]]，表示第一个完成的Future的结果。
 */
def or[T](f1: Future[T])(f2: Future[Option[Nothing]]): Future[Option[T]] =
  Future.firstCompletedOf(Seq(f1.map(Some.apply), f2))

import scala.concurrent.{Future, Await}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global // 导入ExecutionContext
import scala.language.postfixOps // 允许使用 `80.milliseconds` 这样的语法

// 模拟Result类型和异步服务
case class Result(source: String, data: String)

def Web(query: String): Future[Result] = Future {
  Thread.sleep(50) // 模拟网络延迟
  Result("Web", s"Web result for $query")
}

def Image(query: String): Future[Result] = Future {
  Thread.sleep(100) // 模拟网络延迟
  Result("Image", s"Image result for $query")
}

def Video(query: String): Future[Result] = Future {
  Thread.sleep(150) // 模拟网络延迟
  Result("Video", s"Video result for $query")
}

val query = "Scala Async"

// 1. 定义原始的异步请求
val f1 = Web(query)
val f2 = Image(query)
val f3 = Video(query)

// 2. 定义一个全局超时Future
val t = timeout(80.milliseconds)

// 3. 使用Scala的Future API和for-comprehension
// for-comprehension本质上是map/flatMap的语法糖
val resultsUsingFutures: Future[Seq[Result]] = for {
  r1 <- or(f1)(t) // or函数返回Future[Option[Result]]
  r2 <- or(f2)(t)
  r3 <- or(f3)(t)
} yield (r1.toSeq ++ r2.toSeq ++ r3.toSeq) // 将Option[Result]转换为Seq[Result]并拼接

// 4. 或者使用async/await语法（需要引入scala-async库）
// import scala.async.Async.{async, await}
// val resultsUsingAsync: Future[Seq[Result]] = async {
//   val r1 = await(or(f1)(t)) // await会阻塞当前async块直到Future完成
//   val r2 = await(or(f2)(t))
//   val r3 = await(or(f3)(t))
//   r1.toSeq ++ r2.toSeq ++ r3.toSeq
// }

// 示例运行
// val finalResults = Await.result(resultsUsingFutures, 5.seconds)
// println(s"Final Results (using Futures): $finalResults")

// 如果使用async/await，则
// val finalResultsAsync = Await.result(resultsUsingAsync, 5.seconds)
// println(s"Final Results (using Async): $finalResultsAsync")