Scala异步请求的统一超时处理：Future与async/await模式实践

import scala.concurrent.{Future, Promise}
import scala.concurrent.duration.Duration
import scala.concurrent.ExecutionContext.Implicits.global // 或者使用其他ExecutionContext

// 假设存在一个Scheduler，例如Akka的Scheduler，或者自定义的
// 这里我们简化，直接使用ExecutionContext配合一个延迟操作
// 实际生产中可能需要更精确的调度器，如Akka Scheduler
object Scheduler {
  def after(d: Duration)(op: => Unit): Unit = {
    // 这是一个简化的实现，实际可能需要更复杂的调度逻辑
    // 例如使用java.util.concurrent.ScheduledThreadPoolExecutor
    global.execute(() => {
      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]]()
  // 在指定时间后，完成Promise并返回None
  Scheduler.after(d) {
    p.trySuccess(None) // 使用trySuccess避免重复完成
  }
  p.future
}

import scala.concurrent.{Future, Promise}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global // 确保有隐式的ExecutionContext

/**
 * 返回两个Future中首先完成的那一个的结果。
 * 如果f1完成，返回Some(T)；如果f2（通常是超时Future）完成，返回None。
 * @param f1 业务Future
 * @param f2 超时Future (Future[Option[Nothing]])
 * @tparam T 业务Future的结果类型
 * @return 一个Future[Option[T]]，表示业务Future是否在超时前完成
 */
def or[T](f1: Future[T])(f2: Future[Option[Nothing]]): Future[Option[T]] = {
  // 将f1的结果映射为Some(T)，以便与f2的None类型统一
  Future.firstCompletedOf(Seq(f1.map(Some.apply), f2))
}

// 假设Result是某个结果类型
case class Result(source: String, data: String)

// 模拟异步请求函数
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(70) + 20) // 模拟20-90ms延迟
  Result("Image", s"Image result for $query")
}

def Video(query: String): Future[Result] = Future {
  Thread.sleep(scala.util.Random.nextInt(60) + 30) // 模拟30-90ms延迟
  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)

// 使用for-comprehension结合or函数来处理超时
val resultsFuture: Future[Seq[Result]] = for {
  r1 <- or(f1)(t) // r1是Future[Option[Result]]
  r2 <- or(f2)(t) // r2是Future[Option[Result]]
  r3 <- or(f3)(t) // r3是Future[Option[Result]]
} yield {
  // 将Option[Result]转换为Seq[Result]（None变为空Seq，Some(res)变为Seq(res)）
  // 然后合并所有结果
  r1.toSeq ++ r2.toSeq ++ r3.toSeq
}

// 打印结果
resultsFuture.onComplete {
  case scala.util.Success(res) =>
    if (res.isEmpty) {
      println(s"所有请求在${overallTimeout}内全部超时。")
    } else {
      println(s"成功获取 ${res.size} 个结果:")
      res.foreach(r => println(s"  - ${r.source}: ${r.data}"))
    }
  case scala.util.Failure(ex) =>
    println(s"请求过程中发生错误: ${ex.getMessage}")
}
// 为了演示，阻塞等待结果，实际应用中应避免
Thread.sleep(1000)

import scala.async.Async.{async, await}
// 导入其他必要的包和定义（如Result, Web, Image, Video, or, timeout, global等）

val resultsAsync: Future[Seq[Result]] = async {
  // 启动各个异步请求
  val f1 = Web(query)
  val f2 = Image(query)
  val f3 = Video(query)

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

  // 使用await等待每个请求或超时
  val r1 = await(or(f1)(t)) // r1是Option[Result]
  val r2 = await(or(f2)(t)) // r2是Option[Result]
  val r3 = await(or(f3)(t)) // r3是Option[Result]

  // 将Option[Result]转换为Seq[Result]并合并
  r1.toSeq ++ r2.toSeq ++ r3.toSeq
}

// 打印结果（与上面Future API的onComplete相同）
resultsAsync.onComplete {
  case scala.util.Success(res) =>
    if (res.isEmpty) {
      println(s"所有请求在${overallTimeout}内全部超时。")
    } else {
      println(s"成功获取 ${res.size} 个结果:")
      res.foreach(r => println(s"  - ${r.source}: ${r.data}"))
    }
  case scala.util.Failure(ex) =>
    println(s"请求过程中发生错误: ${ex.getMessage}")
}
// 为了演示，阻塞等待结果，实际应用中应避免
Thread.sleep(1000)