在Scala中实现多异步请求的超时控制

import scala.concurrent.{Future, Promise, ExecutionContext}
import scala.concurrent.duration.Duration
import scala.concurrent.ExecutionContext.Implicits.global // 导入全局执行上下文

/**
 * 创建一个在指定持续时间后完成的Future，并返回None。
 * 该Future用于作为超时信号。
 *
 * @param d 超时持续时间。
 * @param ec 隐式的执行上下文，用于调度超时任务。
 * @return 一个在指定时间后成功完成并携带None的Future。
 */
def timeout(d: Duration)(implicit ec: ExecutionContext): Future[Option[Nothing]] = {
  val p = Promise[Option[Nothing]]()
  // 使用执行上下文的scheduleOnce方法在指定时间后完成Promise
  ec.scheduleOnce(d) {
    p.trySuccess(None) // 使用trySuccess避免重复完成Promise
  }
  p.future
}

import scala.concurrent.Future

/**
 * 组合两个Future，返回首先完成的那个Future的结果。
 * 如果f1首先成功完成，则返回Some(f1的结果)；
 * 如果f2(超时Future)首先成功完成，则返回None；
 * 如果f1在f2之前失败，则返回f1的失败。
 *
 * @param f1 实际的任务Future。
 * @param f2 超时信号Future (通常是timeout函数返回的Future[Option[Nothing]])。
 * @param ec 隐式的执行上下文。
 * @tparam T f1 Future的结果类型。
 * @return 一个Future[Option[T]]，表示任务结果或超时。
 */
def or[T](f1: Future[T])(f2: Future[Option[Nothing]])(implicit ec: ExecutionContext): Future[Option[T]] = {
  // 将f1的结果包装成Option[T]，以便与f2的Option[Nothing]类型兼容
  val f1Wrapped: Future[Option[T]] = f1.map(Some.apply)
  // 使用Future.firstCompletedOf来获取首先完成的Future的结果
  Future.firstCompletedOf(Seq(f1Wrapped, f2))
}

import scala.concurrent.Future
import scala.concurrent.duration._ // 导入Duration单位，如80.milliseconds
import scala.concurrent.ExecutionContext.Implicits.global // 导入全局执行上下文
import scala.language.postfixOps // 允许使用后缀操作符，如80.milliseconds

// 假设Result是一个样例类
case class Result(source: String, data: String)

// 模拟异步请求函数
def Web(query: String): Future[Result] = Future {
  Thread.sleep(scala.util.Random.nextInt(50) + 30) // 模拟耗时 30-80ms
  Result("Web", s"Web result for $query")
}

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

def Video(query: String): Future[Result] = Future {
  Thread.sleep(scala.util.Random.nextInt(70) + 10) // 模拟耗时 10-80ms
  Result("Video", s"Video result for $query")
}

// 假设查询字符串
val query = "Scala async"

// 1. 定义原始的异步请求
val fWeb = Web(query)
val fImage = Image(query)
val fVideo = Video(query)

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

// 3. 使用for推导式结合or函数处理每个请求的超时
val resultsFuture: Future[Seq[Result]] = {
  for {
    r1 <- or(fWeb)(globalTimeout)
    r2 <- or(fImage)(globalTimeout)
    r3 <- or(fVideo)(globalTimeout)
  } yield (r1.toSeq ++ r2.toSeq ++ r3.toSeq) // 将Option[Result]转换为Seq[Result]并拼接
}

// 4. 处理最终结果（例如，打印或进一步处理）
resultsFuture.onComplete {
  case scala.util.Success(results) =>
    if (results.isEmpty) {
      println("所有请求均超时或未能成功完成。")
    } else {
      println(s"成功获取 ${results.size} 个结果:")
      results.foreach(println)
    }
  case scala.util.Failure(ex) =>
    println(s"请求处理过程中发生错误: ${ex.getMessage}")
}

// 保持主线程活跃以观察Future结果
Thread.sleep(200) // 等待一段时间让异步操作完成

// build.sbt 示例
libraryDependencies += "org.scala-lang.modules" %% "scala-async" % "1.0.0"

import scala.async.Async.{async, await}
// ... (其他导入和函数定义与上面相同)

val resultsAsyncFuture: Future[Seq[Result]] = async {
  val r1 = await(or(fWeb)(globalTimeout))
  val r2 = await(or(fImage)(globalTimeout))
  val r3 = await(or(fVideo)(globalTimeout))
  // r1, r2, r3 此时是 Option[Result] 类型
  r1.toSeq ++ r2.toSeq ++ r3.toSeq
}

resultsAsyncFuture.onComplete {
  case scala.util.Success(results) =>
    if (results.isEmpty) {
      println("所有请求均超时或未能成功完成 (Async版本)。")
    } else {
      println(s"成功获取 ${results.size} 个结果 (Async版本):")
      results.foreach(println)
    }
  case scala.util.Failure(ex) =>
    println(s"请求处理过程中发生错误 (Async版本): ${ex.getMessage}")
}

Thread.sleep(200) // 等待一段时间让异步操作完成