第二十章 Future 和Promise

在 Scala 中, Future 和 Promise 是用于处理异步操作的强大工具 ;

它们就像一对搭档, 协同工作, 优雅地管理着那些需要花费时间的任务 .

1- 简介

1. Future: 对未来结果的承诺

想象一下, 你正在请求一个需要很长时间才能返回结果的Web服务 ; 与其一直等待, 不如使用 Future ! 它就像一张期票, 承诺在未来某个时刻交付结果 .

• 创建 Future: 你可以使用 Future { /* 耗时操作 */ } 来创建一个 Future 对象; 它会立即返回, 而不会阻塞当前线程 ;
• 回调函数: Future 提供了 onComplete 、onSuccess 和 onFailure 等方法, 允许你在结果可用时执行相应的回调函数 ;
• 组合 Future: 你可以使用 map 、flatMap 和 recover 等方法来组合多个 Future , 从而构建更复杂的异步流程 ;

2. Promise: 兑现 Future 的谎言

Promise 就如同 Future 背后的担保人, 它负责最终完成 Future 并交付结果 ;

• 创建Promise : 你可以使用 Promise[T]() 创建一个 Promise 对象, 其中 T 是结果的类型 ;
• 完成Promise : Promise 对象有一个 success 方法, 可以用来设置 Future 的成功结果, 而 failure 方法则用于设置失败结果 ;
• 链接 Future 和 Promise : Promise 对象创建后会返回一个 Future 对象; 当Promise 完成时, 与其关联的 Future 也会随之完成 ;

3. Future 和 Promise 的关系: 相辅相成

Future 和 Promise 就像一枚硬币的两面:

• Future 代表异步操作的结果, 它是一个只读的对象 ;
• Promise 代表对 Future 的承诺, 它是一个可写的对象 ;

通常情况下, 你会创建一个 Promise对象, 并将其 Future 对象传递给需要异步结果的代码 ; 然后, 在异步操作完成后, 使用 ``Promise的success或failure方法来完成 Future` .

4. 总结

Future 和 Promise 是 Scala 中处理异步编程的强大工具, 它们可以帮你编写更简洁、更高效的并发代码 ;

2- 执行上下文

在 Scala 中, Future 和 Promise 的执行上下文 (ExecutionContext) 决定了异步任务将在哪个线程池中执行 ; 正确理解和管理执行上下文对于编写高效且线程安全的异步代码至关重要 .

1. ExecutionContext 的作用:

• 线程管理: 它维护了一个线程池, 用于执行 Future 中的异步任务 ;
• 资源分配: 它负责为异步任务分配线程资源, 并管理线程的生命周期 ;
• 异常处理: 它提供了默认的异常处理机制, 可以捕获 Future 中未处理的异常 ;

2. 常见的 ExecutionContext :

• global : Scala 提供的全局默认执行上下文, 通常用于简单的异步操作 ;
• fromExecutorService : 可以使用 Java.util.concurrent.ExecutorService 创建自定义的执行上下文, 实现更精细的线程池管理 ;
• fork-join-pool : Scala 2.13版本引入的基于 Frok/Join 框架的执行上下文, 适合处理递归或分治任务 ;

3. 指定 ExecutionContext :

• 隐式参数: 大多数 Future API 都接受一个隐式的 ExecutionContext 参数 ; 如果未指定, 则会使用当前作用域内的隐式值 ;
• 显式传递: 你可以显示地将 ExcutionContext 对象传递给 Future 的方法, 例如 future.onComplete(f)(executionContext) ;

4. 示例:

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{ExecutionContext, Future, Promise}object MyExecutionContextExample {// 自定义执行上下文val myExecutionContext: ExecutionContext = ExecutionContext.fromExecutorService(java.util.concurrent.Executors.newFixedThreadPool(10))def main(args: Array[String]): Unit = {val promise = Promise[String]()val future = promise.future// 在 myExecutionContext 中执行异步任务Future {// 执行耗时操作Thread.sleep(1000)promise.success("异步任务执行完毕")}(myExecutionContext)// 在默认的 global 上下文中处理结果future.onComplete {case scala.util.Success(value) => println(s"结果: $value") // Output: 结果: 异步任务执行完毕case scala.util.Failure(exception) => println(s"异常: ${exception.getMessage}")}Thread.sleep(2000)println("主线程继续执行") // Output: 主线程继续执行}
}

5. 总结:

ExecutionContext 是管理 Future 和 Promise 中异步任务执行的关键组件 ; 理解 ExecutionContext 的作用和使用方法对于编写高效、可靠的并发程序至关重要 ;

6. 注意:

• 避免在 Future 中使用阻塞操作, 因为这会阻塞线程池中的线程, 影响程序性能 ;
• 根据应用场景选择合适的 ExecutionContext , 例如 CPU 密集型任务可以使用固定大小的线程池, I/O 密集型任务可以使用缓存线程池 ;
• 及时关闭自定义的 ExecutionContext , 释放线程资源, 避免资源泄露 ;

3- Future

1. Future: 对未来结果的承诺

Future 就像一张期票, 承诺在未来某个时刻交付结果 ; 它允许你发起一个耗时操作, 并在结果准备好时得到通知, 而无需阻塞当前线程 .

• 非阻塞性: 创建 Future 对象并不会阻塞当前线程 ; 相反, 他会立即返回, 让你可以继续执行其他任务 ;
• 结果获取: 你可以使用回调函数、Await 或组合器来获取 Future 的结果 ;

2. 创建 Future

• Future 伴生对象: 使用 Future {/* 耗时操作 */} 语法, 可以轻松创建一个 Future ; 该操作将在隐式传入的 ExecutionContext 所管理的线程池中执行 ;
• Promise : Promise 提供了一个更灵活的方式来创建和完成 Future ;

3. 处理 Future 结果

• 回调函数: Future 提供了 onComplete 、onSuccess 和 onFailure 等方法, 允许你在结果可用时执行相应的回调函数 ;
• Await : 可以使用 Await.result(future, duration) 阻塞当前线程, 直到 Future 完成并返回结果或超时 ;
• 组合器: Future 提供了一系列组合器, 例如 map 、flatMap 、recover 等, 可以将多个 Future 组合在一起, 构建复杂的异步流程 ;

4. 组合器详解

• map : 对 Future 的成功结果应用一个函数, 并返回一个新的 Future ;
• flatMap : 对 Future 的成功结果应用一个返回 Future 的函数, 并将结果扁平化 ;
• recover : 处理 Future 的失败情况, 并返回一个新的 Future ;
• zip : 将两个 Future 的结果合并成一个元组 ;
• sequence : 将一个 Future 列表转换成一个包含所有结果的 Future ;

5. 异常处理

Future 会自动捕获异步操作中抛出异常; 你可以使用 recover 或 onFailure 来处理这些异常 ;

6. 示例:

import java.io.File
import scala.concurrent.{Await, ExecutionContext, Future}
import scala.io.Sourceobject MyFutureExample {implicit val ec: ExecutionContext = ExecutionContext.globaldef readFileAsync(filePath: String): Future[String] = Future {val source = Source.fromFile(new File(filePath))try {source.getLines().mkString("\n")} finally {source.close()}}def main(args: Array[String]): Unit = {val futureFileContent = readFileAsync("src/main/resources/test.txt") // src/main/resources/test.txtfutureFileContent.onComplete {case scala.util.Success(fileContent) => println(s"文件内容: \n $fileContent")case scala.util.Failure(exception) => println(s"Error reading file: ${exception.getMessage}")}// 让主线程休眠 2 秒，等待异步操作完成Thread.sleep(2000)}
}

输出:

文件内容: 《题西林壁》
宋.苏轼
横看成岭侧成峰，
远近高低各不同。
不识庐山真面目，
只缘身在此山中。

7. 总结

Future 是 Scala中异步编程的基石, 它提供了一种优雅的方式来处理耗时操作, 而不会阻塞主线程 ; 掌握 Future 的创建、结果处理、组合器和异常处理机制, 可以帮助你编写出更响应更健壮的并发程序 ;

4- 阻塞和异常

1. 阻塞的隐患: 异步世界里的绊脚石

Future 的精髓在于非阻塞, 它让我们无需苦苦等待耗时操作完成, 从而提升程序的响应能力 ; 然而, 如果我们在 Future 中使用了阻塞操作, 就会破坏这种非阻塞特性, 将异步执行流拖入同步泥潭 ;

• 线程池枯竭: 阻塞操作会长时间占用线程池中的线程, 导致其他异步任务无法及时执行, 最终可能耗尽线程资源, 使程序陷入瘫痪 ;
• 性能瓶颈: 阻塞操作会阻塞 Future 的执行流程, 降低程序的并发处理能力, 成为性能瓶颈 ;

如何避免阻塞

• 使用异步 API: 尽量使用异步非阻塞的API, 例如异步数据库驱动、异步 HTTP 客户端等 ;
• 将阻塞操作移出 Future : 如果必须使用阻塞操作, 可以将其移出 Future 之外, 或使用专用的线程池来执行 ;
• 使用 blocking 代码快: 对于无法避免的阻塞操作, 可以使用 scala.concurrent.blocking 代码块将其包裹, 告知执行上下文需要额外的线程资源 ;

2. 异常的捕获: 异步世界里的安全网

在异步编程中, 异常处理尤为重要, 因为异常可能发生在任何时间、任何线程 ; 幸运的是, Future 提供了一些机制来帮助我们捕获和处理异常, 防止程序崩溃 ;

• recover 和 recoverWith : 用于捕获 Future 内部抛出的异常, 并根据异常类型进行不同的处理 ; recover 返回一个新的 Future , 而 recoverWith 返回一个新的 Future 或抛出一个新的异常 ;
• onFailure : 注销一个回调函数, 当 Future 执行失败时调用 ;

最佳实践

• 始终处理异常: 不要忽略 Future 中可能抛出的异常 ;
• 使用 try-catch 块: 在 Future 内部使用 try-catch 块来捕获异常 ;
• 根据异常类型进行处理 : 使用 recover 或 recoverWith 根据异常类型进行不同的处理 ;
• 记录异常信息 : 记录异常信息以便于调试和排查问题 ;

5- Promise

1. Promise 与 Future: 相辅相成的异步搭档

• Future : 代表一个异步计算的结果, 它是一个只读对象, 我们无法直接修改它的状态或结构 ;
• promise : 代表一个对 Future 的承诺, 它是一个可以写对象, 我们可以通过它来完成 Future , 决定 Future 是成功还是失败, 以及最终的结果是什么 ;

关系: 每个 Promise 都与一个 Future 相关联, 我们可以通过 Promise 来完成 Future , 一旦 Promise 被完成, 与之关联的 Future 也会随之完成 ;

2. 创建和完成 Promise

• 创建: 使用 Promise[T]() 创建一个新的 Promise 对象, 其中 T 代表 Future 的预期结果类型 ;
• 完成: Promise 提供了三种完成方式 :
  • success(value: T) : 将 Future 标记为成功, 并设置结果值为 value ;
  • failure(exception: Throwable) : 将 Future 标记为失败, 并设置异常信息为 exception ;
  • complete(result: Try[T]) : 根据 result 的值 (Success 或 Failure) 来完成 Future ;

3. 使用 Promise 进行异步编程

Promise 提供了一种更灵活的方式来创建和完成 Future , 特别适合于需要在多个地方协调异步操作的场景 ;

示例: 异步文件下载

import scala.concurrent.{Await, ExecutionContext, Future, Promise}
import java.io.{File, FileOutputStream}
import scala.concurrent.duration.Duration
import scala.io.Sourceobject MyPromiseExample {implicit val ec: ExecutionContext = ExecutionContext.globaldef downloadFile(url: String, filePath: String): Future[Unit] = {val promise = Promise[Unit]()Future {val file = new File(filePath)// 创建文件，如果文件已存在则不创建file.createNewFile()val outputStream = new FileOutputStream(file)try {val source = Source.fromURL(url)source.getLines().foreach(line => outputStream.write((line + "\n").getBytes()))promise.success(()) // 下载成功，完成 Promise} catch {case ex: Exception => promise.failure(ex) // 下载失败，完成 Promise} finally {outputStream.close()}}promise.future}def main(args: Array[String]): Unit = {val downloadFuture = downloadFile("https://raw.githubusercontent.com/******/files/main/poem.txt", "src/main/resources/download_poem_1.txt") // 请将路径替换为您的下载目录// 阻塞 main 函数，直到 downloadFuture 完成Await.result(downloadFuture, Duration.Inf)downloadFuture.onComplete {case scala.util.Success(_) => println("文件下载成功！")case scala.util.Failure(ex) => println(s"文件下载失败：${ex.getMessage}")}}
}

# 下载内容如下: 
《登雀鹤楼》
唐.王之涣
白日依山尽，
黄河入海流。
欲穷千里目，
更上一层楼。

解释:

1. downloadFile 函数使用 Promise 来创建一个 Future[Unit] 对象, 表示文件下载操作 ;
2. 在异步下载任务重, 根据下载结果调用 promise.success() 或 promise.failure() 来完成 Promise , 从而决定 downloadFuture 的最终状态 ;
3. main 函数中, 我们使用 onComplete 回调函数来处理下载结果 ;

4. Promise 的优势

• 更灵活的控制: 可以自由决定 Future 的完成时间和结果 ;
• 多点协调: 可以在多个地方完成 Promise , 实现更复杂的异步流程控制 ;

5. 总结

Promise 为我们提供了掌控 Future 命运的能力, 使得我们可以更灵活地进行异步编程 ; 理解 Promise 与 Future 之间的关系, 以及如何创建、完成和使用 Promise , 可以帮助我们编写出更强大、更易于维护的异步代码 ;

end

end