Scala中如何优雅地实现多个异步请求的超时控制

import scala.concurrent.{Future, Promise}
import scala.concurrent.duration.Duration
import scala.concurrent.ExecutionContext.Implicits.global // 假设存在隐式执行上下文

// 模拟 Scheduler.after，实际项目中请根据所用库引入
// 例如，在Akka项目中，可以使用 akka.pattern.after(d, system.scheduler)(op)
object Scheduler {
  def after(d: Duration)(op: => Unit): Unit = {
    // 实际应用中应避免直接使用 Thread.sleep，而是使用更高级的调度器
    Future {
      Thread.sleep(d.toMillis)
      op
    }
  }
}

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

/**
 * 将一个业务Future与一个超时Future结合，返回两者中先完成的结果。
 * 如果业务Future先完成，结果会被包装成Some(T)；如果超时Future先完成，结果为None。
 * @param f1 业务Future。
 * @param f2 超时Future (通常由timeout函数生成)。
 * @return 一个Future[Option[T]]，表示业务结果或超时信号。
 */
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
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global // 确保有隐式执行上下文
import scala.async.Async.{async, await} // 引入 async/await

// 假设 Result 是一个案例类或某种数据结构
case class Result(source: String, data: String)

// 模拟异步操作，返回Future[Result]
def Web(query: String): Future[Result] = Future {
  Thread.sleep(scala.util.Random.nextInt(50) + 10) // 模拟 10-60ms 延迟
  Result("Web", s"Web result for $query")
}

def Image(query: String): Future[Result] = Future {
  Thread.sleep(scala.util.Random.nextInt(100) + 20) // 模拟 20-120ms 延迟
  Result("Image", s"Image result for $query")
}

def Video(query: String): Future[Result] = Future {
  Thread.sleep(scala.util.Random.nextInt(150) + 30) // 模拟 30-180ms 延迟
  Result("Video", s"Video result for $query")
}

// 定义查询和超时时间
val query = "Scala Async"
val overallTimeout = 80.milliseconds // 设置整体超时时间

// 启动各个异步任务
val f1 = Web(query)
val f2 = Image(query)
val f3 = Video(query)

// 创建一个共享的超时Future
val t = timeout(overallTimeout)

println(s"Starting async requests with an overall timeout of $overallTimeout...")

// 方法一：使用 Scala 的 Future API (for-comprehension)
// 这种方式将多个带超时的Future组合起来，收集所有在超时前完成的结果
val resultsUsingFuture: Future[Seq[Result]] = for {
  r1 <- or(f1)(t) // Web请求或超时
  r2 <- or(f2)(t) // Image请求或超时
  r3 <- or(f3)(t) // Video请求或超时
} yield (r1.toSeq ++ r2.toSeq ++ r3.toSeq) // 将Option[Result]转换为Seq[Result]并拼接

// 方法二：使用 async/await 语法
// 提供了更接近同步代码的编写体验，但本质上是相同的异步逻辑
val resultsUsingAsync: Future[Seq[Result]] = async {
  val r1 = await(or(f1)(t)) // 等待Web请求或超时
  val r2 = await(or(f2)(t)) // 等待Image请求或超时
  val r3 = await(or(f3)(t)) // 等待Video请求或超时
  r1.toSeq ++ r2.toSeq ++ r3.toSeq // 收集结果
}

// 打印结果 (示例)
resultsUsingAsync.onComplete {
  case scala.util.Success(res) =>
    println(s"Collected results within timeout: ${res.map(_.source).mkString(", ")}")
    if (res.isEmpty) println("No results collected within the timeout period.")
  case scala.util.Failure(ex) =>
    println(s"An error occurred: ${ex.