Scala中Future.sequence超时后如何定位未完成Future及添加元数据？

1. 找出未按时完成的Future #

有两种实用方案可以定位超时的任务：

方式一：给单个Future绑定独立超时与元数据 #

不要等到Future.sequence统一超时，而是给每个Future单独包装超时逻辑并绑定任务标识（如订单ID、服务名称）。单个任务超时会直接抛出带具体信息的异常，精准定位问题。

示例代码：

import scala.concurrent.{Future, Await}
import scala.concurrent.duration._
import scala.util.control.NonFatal

// 包装函数：为Future添加超时和任务元数据
def withTaskMetaAndTimeout[T](
  future: Future[T],
  taskMeta: String,
  timeout: FiniteDuration
): Future[T] = {
  val timeoutSignal = Future {
    Thread.sleep(timeout.toMillis)
    throw new TimeoutException(s"任务超时: $taskMeta")
  }
  // 取第一个完成的结果（正常返回或超时）
  Future.firstCompletedOf(Seq(future, timeoutSignal))
}

// 原始Future列表与对应的任务元数据
val taskMetas = List("轮询支付服务[订单1001]", "轮询物流服务[运单2002]", "轮询库存服务[商品3003]")
val listOfFutures: List[Future[String]] = // 你的原始Future集合

// 为每个Future绑定元数据和超时
val wrappedFutures = listOfFutures.zip(taskMetas).map {
  case (f, meta) => withTaskMetaAndTimeout(f, meta, 10.minutes)
}

try {
  val results = Await.result(Future.sequence(wrappedFutures), 10.minutes)
} catch {
  case te: TimeoutException =>
    println(te.getMessage) // 直接输出超时任务的具体信息
  case NonFatal(e) =>
    e.printStackTrace()
}

方式二：整体超时后检查未完成任务 #

如果必须保留整体超时逻辑，可以提前将每个Future与元数据关联，超时后遍历检查哪些Future仍未完成：

import scala.concurrent.{Future, Await}
import scala.concurrent.duration._

// 关联Future和任务元数据
val taskMetas = List("订单1001", "运单2002", "商品3003")
val futureMetaPairs = listOfFutures.zip(taskMetas)

val allFutures = Future.sequence(listOfFutures)

try {
  val results = Await.result(allFutures, 10.minutes)
} catch {
  case _: TimeoutException =>
    // 筛选出未完成的任务
    val incompleteTasks = futureMetaPairs.collect {
      case (f, meta) if !f.isCompleted => meta
    }
    println(s"超时后未完成的任务: ${incompleteTasks.mkString(", ")}")
}

注意：isCompleted是瞬时状态检查，可能在执行检查时部分Future刚好完成，但仍能定位到绝大多数超时任务。

2. 为Future添加额外元数据，展示任务信息 #

核心是将任务元数据（如订单ID、服务名称、请求参数）与Future绑定，在异常或状态检查时携带这些信息：

自定义异常携带元数据 #

定义专属的超时异常类，将任务元数据作为异常字段，方便排查时获取上下文：

case class TaskTimeoutException(taskMeta: String, message: String) extends Exception(message)

// 改造包装函数
def withTaskMetaAndTimeout[T](
  future: Future[T],
  taskMeta: String,
  timeout: FiniteDuration
): Future[T] = {
  val timeoutSignal = Future.failed(
    TaskTimeoutException(taskMeta, s"任务超时（10分钟）: $taskMeta")
  )
  Future.firstCompletedOf(Seq(future, timeoutSignal))
}

// 捕获异常时提取元数据
try {
  val results = Await.result(Future.sequence(wrappedFutures), 10.minutes)
} catch {
  case tte: TaskTimeoutException =>
    println(s"任务[${tte.taskMeta}]超时，原因：${tte.getMessage}")
    // 可将元数据写入日志或监控系统
}

用自定义类包装结果与元数据 #

如果需要同时保留正常结果和元数据，可以用自定义case class包装：

case class TaskResult[T](taskMeta: String, result: T)

// 包装Future，返回带元数据的结果
val wrappedFutures = listOfFutures.zip(taskMetas).map {
  case (f, meta) => f.map(res => TaskResult(meta, res))
}

// 超时后通过元数据定位未完成任务
try {
  val results = Await.result(Future.sequence(wrappedFutures), 10.minutes)
} catch {
  case _: TimeoutException =>
    val incomplete = futureMetaPairs.filter { case (f, _) => !f.isCompleted }
    incomplete.foreach { case (_, meta) =>
      println(s"任务[$meta]未完成，可能因外部服务响应缓慢")
    }
}

内容的提问来源于stack exchange，提问作者Capacytron