Scala 3中Tuple与Future元组互转及超时处理问题

问题描述 #

我需要实现一个方法，接收任意类型的Future元组，返回已完成Future的结果元组，同时为Future的完成设置时间限制。尝试用Tuple.Map匹配类型编写代码时遇到编译错误，修改后虽能运行，但存在两个问题：

• 运行时类型检查警告
• 调用时必须手动指定类型标注，使用不便

原代码 #

def getAll[T <: Tuple](futures: Tuple.Map[T, Future])(timeout: Long, units: TimeUnit): T = futures match
    case e: EmptyTuple => EmptyTuple.asInstanceOf[T]
    case fs: (fh *: ft) => // 此处认为fh应为Future类型
        val start = System.currentTimeMillis()
        val vh = fs.head.asInstanceOf[fh].get(timeout, units)
        val elapsed = System.currentTimeMillis() - start
        val remaining = TimeUnit.MILLISECONDS.convert(timeout, units) - elapsed

        vh *: getAll(fs.tail)(remaining)

编译错误 #

value get is not a member of fh

修改后代码 #

def getAll[T <: Tuple](futures: Tuple.Map[T, Future])(timeout: Long, units: TimeUnit): T = futures match
    case _: EmptyTuple => EmptyTuple.asInstanceOf[T]
    case fs: Tuple.Map[fh *: ft, Future] =>
        val start = System.nanoTime()
        val vh = fs.head.asInstanceOf[Future[fh]].get(timeout, units)
        val elapsed = System.nanoTime() - start
        val remaining = TimeUnit.NANOSECONDS.convert(timeout, units) - elapsed

        (vh *: getAll(fs.tail)(remaining, TimeUnit.NANOSECONDS)).asInstanceOf[T]

运行时警告 #

scala.Tuple.Map[ft, java.util.concurrent.Future]
) @ft @fh cannot be checked at runtime

调用示例（需手动指定类型） #

getAll[(String, String)]((f1, f2))(to, TimeUnit.SECONDS) // 必须指定类型标注才能编译

解决方案 #

通过Scala的类型类和编译时类型推导，可以彻底解决上述问题，无需运行时类型转换、消除警告，同时让编译器自动推导返回类型。

正确实现代码 #

import java.util.concurrent.{Future, TimeUnit}
import scala.compiletime.ops.tuple.IsEmpty
import scala.deriving.Mirror

trait GetAll[T <: Tuple] {
  type Out <: Tuple
  def apply(futures: T, timeoutNs: Long): Out
}

object GetAll {
  type Aux[T <: Tuple, O <: Tuple] = GetAll[T] { type Out = O }

  // 空元组的处理实例
  given getAllEmpty: Aux[EmptyTuple, EmptyTuple] with {
    def apply(futures: EmptyTuple, timeoutNs: Long): EmptyTuple = EmptyTuple
  }

  // 非空元组的递归处理实例
  given getAllCons[H, T <: Tuple](using next: GetAll[T]): Aux[Future[H] *: T, H *: next.Out] with {
    def apply(futures: Future[H] *: T, timeoutNs: Long): H *: next.Out =
      val start = System.nanoTime()
      // 处理头部Future，获取结果
      val headResult = futures.head.get(timeoutNs, TimeUnit.NANOSECONDS)
      // 计算剩余超时时间
      val elapsed = System.nanoTime() - start
      val remaining = timeoutNs - elapsed
      // 递归处理尾部元组，拼接结果
      headResult *: next(futures.tail, remaining)
  }

  // 对外暴露的调用方法
  def getAll[T <: Tuple](futures: T)(timeout: Long, units: TimeUnit)(using ga: GetAll[T]): ga.Out =
    val timeoutNs = TimeUnit.NANOSECONDS.convert(timeout, units)
    ga(futures, timeoutNs)
}

使用示例 #

import GetAll.getAll

// 定义测试用的Future
val f1: Future[String] = java.util.concurrent.CompletableFuture.completedFuture("hello")
val f2: Future[Int] = java.util.concurrent.CompletableFuture.completedFuture(42)

// 无需手动指定类型，编译器自动推导返回类型为(String, Int)
val result = getAll((f1, f2))(5, TimeUnit.SECONDS)
// result: (String, Int) = ("hello", 42)

方案说明 #

1. 编译时类型推导：通过GetAll类型类，为元组的空情况和非空情况分别提供实现，让编译器在编译阶段就确定每个Future的类型及返回元组的结构，完全避免运行时类型转换。
2. 消除运行时警告：所有类型逻辑都在编译期处理，不需要asInstanceOf强制转换，自然消除了类型检查警告。
3. 自动类型推导：调用时无需手动指定类型参数，编译器会根据传入的Future元组自动推导结果元组的类型，使用更便捷。
4. 超时时间管理：每次处理头部Future后，计算剩余超时时间并传递给尾部元组的处理，严格遵守整体超时限制。

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