搜索菜单
当前位置: 首页 > Scala > Scala 异步编程入门：Future、Promise、超时控制与阻塞等待

Scala 异步编程入门：Future、Promise、超时控制与阻塞等待

林舟
林舟
发布时间：2026-08-23 | 34人浏览过
原创

前言

如果你刚从 C++ 或同步式代码切到 Scala，最容易卡住的地方通常不是语法，而是“任务什么时候开始、结果什么时候可用、失败怎么处理、超时怎么兜底”。本文按这个思路，把 Scala 异步编程里最常见的 `Future`、`Promise`、结果组合和阻塞等待串起来讲一遍。

如果你刚从 C++ 或同步式代码切到 Scala，最容易卡住的地方通常不是语法，而是“任务什么时候开始、结果什么时候可用、失败怎么处理、超时怎么兜底”。本文按这个思路，把 Scala 异步编程里最常见的 `Future`、`Promise`、结果组合和阻塞等待串起来讲一遍。

文中的命令、定义、代码与输出示例均保留原始事实，你可以一边对照示例，一边判断哪些场景适合回调、哪些更适合组合，什么时候又该显式阻塞等待结果。

Future 是什么，最基本怎么用

本文更适合有 C++ 基础、并且已经熟悉 Scala 基本概念的读者。先从 `Future` 本身看起，它的基本概念和 C++ 的 std::Future 类似，定义如下：

展示 Scala Future 从创建、执行到回调处理结果的流程信息图
Future 创建到回调的执行路径用流程图梳理 `Future` 的创建、执行线程与成功回调触发关系。
trait Future[T]

Future 还提供了 apply 方法，声明方式如下：

def apply[T](b: => T)(implicit e: ExecutionContext): Future[T]

其中 => T 的定义可以参考这篇文章。上面的写法还涉及柯理化与隐式参数，想补这部分背景可以继续看这篇文章。

先看一个最基础的例子：

package TestFuturePromise

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future

object TestFuture {
  def main(args: Array[String]): Unit = {
    val future = Future {  // Future[Unit] 类型的
      Thread.sleep(1000)
      val tid = Thread.currentThread().getName
      println(s"future finished in $tid")
    }

    while (!future.isCompleted) {
      println("main thread wait for future")
      Thread.sleep(200)
    }
  }

  // 模拟下载
  def downLoadUrl(url: String): String = {
    val t = 3
    Thread.sleep(1000 * t)
    val tid = Thread.currentThread().getName
    println(s"finish download after $t s from thread $tid")
    "hello world"
  }
}

输出结果：

main thread wait for future
main thread wait for future
main thread wait for future
main thread wait for future
main thread wait for future
future finished in scala-execution-context-global-13

这个例子最直观地说明了一点：主线程并没有立即拿到结果，而是在不断检查 future.isCompleted，直到异步任务执行结束。

Future 完成后，怎么拿结果

和 C++ 常见写法不同，Scala 支持直接在 Future 上附加回调，最常见的是使用 foreach。先看接口定义：

def foreach[U](f: scala.Function1[T, U])(implicit executor: scala.concurrent.ExecutionContext): scala.Unit

回调函数接收的是 `Future` 成功完成后的结果，同时还需要一个 `ExecutionContext`。下面直接在前面的示例上做修改：

object TestFuture {
  def main(args: Array[String]): Unit = {
    val future = Future {  // Future[String] 类型的
      Thread.sleep(1000)
      val tid = Thread.currentThread().getName
      println(s"future finished in $tid")
      "hello future" // 这里新增返回值
    }

    // 增加回调函数
    future.foreach(s => {
      val tid = Thread.currentThread().getName
      println(s"callback from $tid get content $s")
    })

    while (!future.isCompleted) {
      println("main thread wait for future")
      Thread.sleep(200)
    }
  }
}

foreach 只会在 `Future` 成功时触发，这一点很关键。原文里也提到，它和 onSuccess 的语义接近；如果是失败情况，可以使用 onFailed。当你明确知道任务可能抛异常时，更常见的做法是直接使用 onComplete，再通过 match 处理 Try[T]。关于 `Try`，可以参考这篇文章。

Future 怎么组合：从 map 开始

仅仅“启动一个异步任务并等它结束”还不够，真实业务里更常见的是把异步结果继续变换。这里最基础的工具就是 map。

原文对它的描述是：map 用于转化 Future，可以理解为链接多个 Future。示例如下：

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.util.{Failure, Success}

object TestFuture {
  def main(args: Array[String]): Unit = {
    val result = Future[Int] {
      1
    }.map {
      num => {
        println("int to string map")
        num + "234"
      }
    }

    result.onComplete {
      case Success(num) => println(num)
      case Failure(e) => println(e.getMessage)
    }

    Thread.sleep(10000)
  }
}

最终输出：

int to string map
1234

这里有个很实用的点：在 map 里不需要显式返回 Future，只要返回目标值即可，Scala 会自动包装成对应的 Future。也就是说，`Future[Int]` 经过这次 `map` 之后，变成了 `Future[String]`。

除了 map，原文还提到 filter、flatMap 等其他组合方式，完整说明可以直接参考 Scala 官方文档。

Promise 的作用：把结果生产者和消费者拆开

如果说 `Future` 更像“已经安排出去的一次异步计算”，那 `Promise` 的重点就在于：谁来消费结果，和谁来填充结果，可以分开。

展示 Promise 如何分离结果生产者与消费者，以及 success 和 failure 两条完成路径的信息图
Promise 如何手动完成 Future把 `Promise` 看成“手动完成 Future 的入口”会更容易理解：调用方只看到。

原文里把这个差异解释得很清楚。使用 `Future` 时，我们往往在构造时就把计算逻辑写进去；而使用 `Promise` 时，提供方先把一个 `Future` 暴露给调用者，自己则在合适的时机调用 `success` 或 `failure` 去完成它。这种做法在需要手动桥接回调、事件或底层异步 API 时非常常见。

示例代码如下：

package TestFuturePromise

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future, Promise}
import scala.util.{Failure, Random, Success}

object TestFuture {
  def main(args: Array[String]): Unit = {
    val fut = downLoad("www.foo.com", tag = true)
    fut.onComplete {
      case Success(txt) => println(txt)
      case Failure(e) => println(e.getMessage)
    }

    val fut1 = downLoad("www.foo1.com", tag = false)
    fut1.onComplete {
      case Success(txt) => println(txt)
      case Failure(e) => println(e.getMessage)
    }

    println("finish add callback")
    Thread.sleep(5 * 1000)
    println("main thread finished")
  }

  def downLoad(url: String, tag: Boolean): Future[String] = {
    val promise = Promise[String]
    // 模拟异步下载
    global.execute(() => {
      val t = Random.between(1, 4)
      Thread.sleep(t * 1000)
      val tid = Thread.currentThread().getName
      if (tag) {
        println(s"$tid download from [$url] takes $t s")
        promise.success("hello world")
      } else {
        println(s"$tid download from [$url] [FAILED] takes $t s")
        promise.failure(new Exception("download failed"))
      }
    })
    promise.future
  }
}

输出结果：

finish add callback
scala-execution-context-global-14 download from [www.foo1.com] [FAILED] takes 3 s
download failed
scala-execution-context-global-13 download from [www.foo.com] takes 4 s
hello world
main thread finished

这个例子里，调用方只拿到 `Future[String]`，并用 `onComplete` 注册成功或失败回调；真正决定何时返回结果、返回成功还是失败的，是 `downLoad` 方法内部的 `Promise[String]`。这就是 `Promise` 最核心的价值：把“结果出口”和“结果入口”拆开。

没有内建 timeout 时，怎么做超时和抢先返回

原文特别指出，`Future` 本身没有 `timeout` 方法。如果业务要求“超过某个时间就立刻返回”，可以借助 Java 的 `Timer` 来包装一个超时 `Future`。

展示 timeout 与自定义 or 方法如何让两个 Future 竞争完成并返回先到结果的信息图
timeout 与 or 的抢先返回逻辑这一节的重点不是定时器本身，而是借助 `Promise.tryComplete` 把“超时。

先实现一个最简单的 timeout

package TestFuturePromise

import java.util.{Timer, TimerTask}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future, Promise}

object TestFuture {
  def main(args: Array[String]): Unit = {
    timeout(1000).foreach(_ => println("timer Future finished"))
    Thread.sleep(2000)
  }

  def timeout(t: Int): Future[Unit] = {
    val timer = new Timer(true)
    val p = Promise[Unit]
    timer.schedule(new TimerTask {
      override def run(): Unit = {
        p.success(())  // 超时了，此时立刻设置返回值
        timer.cancel()
      }
    }, t)
    p.future
  }
}

这个版本的作用很简单：等到 `t` 毫秒后，把 `Promise[Unit]` 标记成功，从而让返回的 `Future[Unit]` 结束。单独看它意义有限，但它为后面的“谁先完成就用谁”打下了基础。

再把 timeout 和业务 Future 组合起来

更实用的场景是：给任意 `Future[T]` 增加超时控制，如果超时先到，就返回超时结果；如果业务任务先结束，就直接返回业务结果。原文通过自定义 `or` 方法实现了这个思路：

package TestFuturePromise

import java.util.{Timer, TimerTask}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future, Promise}
import scala.util.{Failure, Success}

object TestFuture {
  def main(args: Array[String]): Unit = {
    // 利用 map，转换成 Future[String]类型，并返回超时的处理结果
    val f = timeout(1000).map(_ => "timeout") or Future {
      Thread.sleep(8)
      println("or Future callback")
      "hello timeout"
    }

    f.onComplete {
      case Success(txt) => println(txt)
      case Failure(e) => println(e.getMessage)
    }
    Thread.sleep(2000)
  }

  def timeout(t: Int): Future[Unit] = {
    val timer = new Timer(true)
    val p = Promise[Unit]
    timer.schedule(new TimerTask {
      override def run(): Unit = {
        p.success(())
        timer.cancel()
      }
    }, t)
    p.future
  }

  implicit class FutureOp[T](val self: Future[T]) {
    def or(that: Future[T]): Future[T] = {
      val p = Promise[T]
      // tryComplete 只会成功调用一次
      self.onComplete(x => p.tryComplete(x))
      that.onComplete(y => p.tryComplete(y))
      p.future
    }
  }
}

这里最值得注意的是 `or` 的实现。它内部再创建一个 `Promise[T]`，然后同时为 `self` 和 `that` 注册 `onComplete` 回调，两个回调都尝试调用 `p.tryComplete(...)`。由于 `tryComplete` 只会真正成功一次，因此谁先完成，最终返回的就是谁的结果。

原文这里还用到了 `implicit class` 扩展方法，相关语法可以参考这篇文章。

什么时候该阻塞：Await 的基本用法

虽然 `Future` 和 `Promise` 的核心目的是非阻塞，但在某些边界场景里，显式阻塞仍然有用。比如脚本式入口、测试代码，或者必须在当前线程拿到最终结果的地方，可以使用 `Await`。

示例如下：

package TestFuturePromise

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration.DurationInt
import scala.concurrent.{Await, Future}

object TestFuture {
  def main(args: Array[String]): Unit = {
    val fut = Future {
      Thread.sleep(1000)
      println("future weak up")
      "hello Await"
    }
    Await.result(fut, 1200.seconds)
    println(fut.value)
  }
}

这段代码的重点很直接：Await.result(fut, 1200.seconds) 会阻塞当前线程，直到 `fut` 完成或超时。对于理解异步模型来说，它能帮助你区分两类需求：一类是“任务完成后通知我”，另一类是“我必须现在就拿到结果”。前者更适合回调和组合，后者才考虑阻塞。

怎么选：把几种常见写法放到同一个判断框架里

把全文串起来看，可以得到一套很实用的使用顺序：

  • 只想启动异步任务并在完成后处理结果，用 Future 加 foreach / onComplete。
  • 需要把异步结果继续变换、串联后续处理，用 map、flatMap 等组合方式。
  • 结果的生产时机不在当前计算块里，或者需要手动控制成功与失败，用 Promise。
  • 需要“超时就返回”或“两个 Future 谁先完成用谁”，可以用 Promise 再包一层竞争逻辑。
  • 只有在确实必须同步取值时，才用 Await 做阻塞等待。

如果你已经有 C++ 的 `std::Future` 背景，最需要补的其实不是“怎么开异步任务”，而是 Scala 这套围绕 `Future` 提供的回调、组合和手动完成能力。理解了这几块，再看更复杂的异步控制就会顺畅很多。

相关文章

相关标签: Scala 并发编程

本站声明

本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。

作者最新文章

全部内容

Python 相关产品和应用

全部下载
Python Selenium
Selenium

Selenium

4.51.0 · Windows

用于浏览器自动化和 Web 测试的 Python 工具。
Python 机器学习
scikit-learn

scikit-learn

1.9.1 · Windows

常用 Python 机器学习算法与模型评估工具库。
Python 代码质量
Black

Black

26.10.1 · Windows

强调一致性的 Python 自动代码格式化工具。
Python 代码质量
Ruff

Ruff

0.17.1 · macOS

由 Rust 编写的高速 Python 代码检查和格式化工具。
Python pytest
pytest

pytest

9.1.1 · Windows

简洁而强大的 Python 自动化测试框架。
Python FastAPI API
FastAPI

FastAPI

0.146.0 · Windows

高性能 Python API 框架，支持类型提示和自动接口文档。
Python Flask Web开发
Flask

Flask

3.1.3 · Windows

轻量灵活的 Python Web 应用框架。
Python Django Web开发
Django

Django

6.1.2 · Windows

成熟的 Python Web 框架，适合构建内容站点和业务系统。
Python Spyder
Spyder

Spyder

6.1.7 · Windows

面向科学计算与数据分析的 Python 集成开发环境。
Python SciPy
SciPy

SciPy

1.18.1 · Windows

面向科学、工程和技术计算的 Python 算法库。
Python Pandas 数据分析
pandas

pandas

3.0.6 · Windows

Python 数据分析库，提供 DataFrame 与丰富的数据处理能力。
Python NumPy
NumPy

NumPy

2.5.4 · Windows

Python 科学计算基础库，提供高性能多维数组与数值运算。

文章分类

专题列表

ipython.cn 学习网
人工智能 Java AI行业服务