在任何并发性应用程序中,异步事件处理都至关重要。无论事件的来源是什么(不同的计算任务、I/O 操作或与外部系统的交互),您的代码都必须跟踪事件,协调为响应它们而执行的操作。应用程序可以采用两种基本方法之一来实现异步事件处理:
- 阻塞:一个等待事件的协调线程。
- 非阻塞:事件向应用程序生成某种形式的通知,而没有线程显式等待它。
在 “JVM 并发性:阻塞还是非阻塞?” 中,可以了解 Java™ 8 中处理异步事件的方式,并使用基于CompletableFuture
类的阻塞和非阻塞技术。本教程将展示 Scala 中一些异步事件处理选项,首先会介绍一个简单的阻塞版本,然后介绍一些非阻塞选项。最后,您会看到 async
/ await
结构如何将简单的阻塞代码转换为非阻塞执行形式。(从作者的 GitHub 存储库中可以获得 完整的示例代码。)
合成事件
scala.concurrent.Promise
和 scala.concurrent.Future
类为 Scala 开发人员提供了一些与 Java 8 开发人员的 CompletableFuture
使用方式类似的选项。具体地讲,Future
同时提供了阻塞和非阻塞的事件完成方式。但是,尽管在此级别上很相似,但用于处理两种 future 的技术是不同的。
在本节中,您将看到处理 Future
所表示的事件的阻塞和非阻塞方法示例。本教程使用了与上一个教程相同的并发任务设置。在深入探讨代码之前,我将回顾一下这些设置。
任务和排序
在一个特定操作中,应用程序通常必须执行多个处理步骤。例如,在向用户返回结果之前,Web 应用程序可能需要:
- 在一个数据库中查找用户的信息
- 使用查找到的信息来执行 Web 服务调用,并执行另一次数据库查询。
- 根据从前两个操作中获得的结果来执行数据库更新。
图 1 演示了这种结构类型。
图 1. 应用程序任务流
图 1 将处理过程分解为 4 个不同的任务,它们通过表示顺序依赖关系的箭头相连接。任务 1 可以直接执行,任务 2 和任务 3 都在任务 1 完成后执行,任务 4 在任务 2 和任务 3 都完成后执行。
建模异步事件
在真实的系统中,异步事件的来源一般是并行计算或一种形式的 I/O 操作。但是,使用简单的时间延迟来建模这种系统会更容易一些,这也是这里所采用的方法。清单 1 显示了我用于生成事件的基本的赋时事件 (timed-event) 代码,这些事件采用了已完成的 Future
格式。
清单 1. 赋时事件代码
import java.util.Timer import java.util.TimerTask import scala.concurrent._ object TimedEvent { val timer = new Timer /** Return a Future which completes successfully with the supplied value after secs seconds. */ def delayedSuccess[T](secs: Int, value: T): Future[T] = { val result = Promise[T] timer.schedule(new TimerTask() { def run() = { result.success(value) } }, secs * 1000) result.future } /** Return a Future which completes failing with an IllegalArgumentException after secs * seconds. */ def delayedFailure(secs: Int, msg: String): Future[Int] = { val result = Promise[Int] timer.schedule(new TimerTask() { def run() = { result.failure(new IllegalArgumentException(msg)) } }, secs * 1000) result.future }
像上一期的 Java 代码一样,清单 1 中的 Scala 代码使用一个 java.util.Timer
来安排 java.util.TimerTask
在一个延迟之后执行。每个 TimerTask
在运行时完成一个有关联的 future。delayedSuccess
函数定制了一个任务,在运行时成功完成一个 Scala Future[T]
,然后将该 future 返回给调用方。delayedSuccess
函数返回相同类型的 future,但使用了一个在完成 future 时发生IllegalArgumentException
异常的失败任务。
清单 2 展示了如何使用 清单 1 中的代码创建 Future[Int]
格式的事件,使之与 图 1 中的 4 个任务相匹配。(此代码来自示例代码中的AsyncHappy
类。)
清单 2. 示例任务的事件
// task definitions def task1(input: Int) = TimedEvent.delayedSuccess(1, input + 1) def task2(input: Int) = TimedEvent.delayedSuccess(2, input + 2) def task3(input: Int) = TimedEvent.delayedSuccess(3, input + 3) def task4(input: Int) = TimedEvent.delayedSuccess(1, input + 4)
清单 2 中 4 个任务方法中的每一个都为该任务的完成时刻使用了特定的延迟值:task1
为 1 秒,task2
为 2 秒,task3
为 3 秒,task4
重新变为 1 秒。每个任务还接受一个输入值,是该输入加上任务编号作为 future 的(最终)结果值。这些方法都使用了 future 的成功形式;稍后您会看到一些使用失败形式的例子。
这些任务要求您按 图 1 中所示的顺序运行它们,向每个任务传递上一个任务返回的结果值(或者对于 task4
,传递前两个任务结果的和)。如果中间两个任务同时执行,总的执行时间大约为 5 秒(1 秒 + (2 秒、3 秒中的最大值)+ 1 秒。如果 task1
的输入为 1,那么结果为 2。如果该结果被传递给 task2
和 task3
,那么结果将为 4 和 5。如果这两个结果的和 (9) 被作为输入传递给 task4
,那么最终结果将为 13。
阻塞等待
在设定好操作环境之后,是时候来查看 Scala 如何处理事件的完成情况了。与上一期的 Java 代码中一样,协调 4 个任务的执行的最简单的方法是使用阻塞等待:主要线程等待每个任务依次完成。清单 3(同样来自示例代码中的 AsyncHappy
类)给出了此方法。
清单 3. 阻塞等待任务执行
def runBlocking() = { val v1 = Await.result(task1(1), Duration.Inf) val future2 = task2(v1) val future3 = task3(v1) val v2 = Await.result(future2, Duration.Inf) val v3 = Await.result(future3, Duration.Inf) val v4 = Await.result(task4(v2 + v3), Duration.Inf) val result = Promise[Int] result.success(v4) result.future }
清单 3 使用 Scala scala.concurrent.Await
对象的 result()
方法来完成阻塞等待。该代码首先等待 task1
的结果,然后同时创建task2
和 task3
future,并等待两个任务依次返回 future,最后等待 task4
的结果。最后 3 行(创建和设置 result
)使得该方法能够返回一个 Future[Int]
。返回该 future,让此方法与我接下来展示的非阻塞形式一致,但该 future 将在该方法返回之前完成。
组合 future
清单 4(同样来自示例代码中的 AsyncHappy
类)展示了一种将 future 联系在一起的方式,以便按正确顺序并使用正确的依赖关系执行任务,而不使用阻塞。
清单 4. 使用 onSuccess()
处理事件的完成
def runOnSuccess() = { val result = Promise[Int] task1(1).onSuccess(v => v match { case v1 => { val a = task2(v1) val b = task3(v1) a.onSuccess(v => v match { case v2 => b.onSuccess(v => v match { case v3 => task4(v2 + v3).onSuccess(v4 => v4 match { case x => result.success(x) }) }) }) } }) result.future }
清单 4 代码使用 onSuccess()
方法将一个函数(技术上讲是一个部分函数,因为它仅处理成功完成的情况)设置为在每个 future 完成时返回。因为 onSuccess()
调用是嵌套式的,所以它们将按顺序执行(即使 future 未完全按顺序完成)。
清单 4 的代码比较容易理解,但很冗长。清单 5 展示了一种使用 flatMap()
方法处理这种情况的更简单的方法。
清单 5. 使用 flatMap()
处理事件的完成
def runFlatMap() = { task1(1) flatMap {v1 => val a = task2(v1) val b = task3(v1) a flatMap { v2 => b flatMap { v3 => task4(v2 + v3) }} } }
清单 5 中的代码实际上执行了与 清单 4 相同的事情,但 清单 5 使用了 flatMap()
方法从每个 future 中提取单一结果值。使用flatMap()
消除了 清单 4 中所需的 match
/ case
结构,提供了一种更简洁的格式,但采用了同样的逐步执行路线。
试用示例
示例代码使用了一个 Scala App
来依次运行事件代码的每个版本,并确保完成事件(约 5 秒)和结果 (13) 是正确的。您可以使用 Maven 从命令行运行此代码,如清单 6 所示(删除了无关的 Maven 输出):
清单 6. 运行事件代码
dennis@linux-9qea:~/devworks/scala4/code> mvn scala:run -Dlauncher=happypath ... [INFO] launcher 'happypath' selected => com.sosnoski.concur.article4.AsyncHappy Starting runBlocking runBlocking returned 13 in 5029 ms. Starting runOnSuccess runOnSuccess returned 13 in 5011 ms. Starting runFlatMap runFlatMap returned 13 in 5002 ms. �
不顺利的道路
目前为止,您看到了以 future 形式协调事件的代码,这些代码总是能够成功完成。在真实应用程序中,不能寄希望于事情总是这么顺利。处理任务过程中可能会出现问题,而且在 JVM 语言术语中,这些问题通常表示为 Throwable
。
更改 清单 2 中的任务定义很容易,只需使用 delayedFailure()
代替 delayedSuccess()
方法,如这里的 task4
所示:
def task4(input: Int) = TimedEvent.delayedFailure(1, "This won't work!")
如果运行仅将 task4
修改为完成时抛出异常的 清单 3,那么您会得到 task4
上的 Await.result()
调用所抛出的预期的IllegalArgumentException
。如果在 runBlocking()
方法中没有捕获该问题,该异常会在调用链中一直传递,直到最终捕获问题(如果未捕获问题,则会终止线程)。幸运的是,修改该代码很容易,因此,如果任何任务完成时抛出异常,该异常会通过返回的 future 传递给调用方来处理。清单 7 展示了这一更改。
清单 7. 具有异常的阻塞等待
def runBlocking() = { val result = Promise[Int] try { val v1 = Await.result(task1(1), Duration.Inf) val future2 = task2(v1) val future3 = task3(v1) val v2 = Await.result(future2, Duration.Inf) val v3 = Await.result(future3, Duration.Inf) val v4 = Await.result(task4(v2 + v3), Duration.Inf) result.success(v4) } catch { case t: Throwable => result.failure(t) } result.future }
清单 7 非常浅显易懂,最初的代码包装在一个 try
/catch
中,catch
在返回的 future 完成时传回异常。此方法稍微复杂一些,但任何 Scala 开发人员应该仍然很容易理解它。
那么,清单 4 和清单 5 中的事件处理代码的非阻塞变形是怎样的?从名称可以看出,清单 4 中使用的 onSuccess()
方法仅 适用于 future 的成功完成类型。如果想要同时处理成功和失败完成类型,则必须使用 onComplete()
方法,检查哪种完成例行适用。清单 8 展示了此技术如何用在事件处理代码中。
清单 8. 成功和失败的 onComplete()
处理
def runOnComplete() = { val result = Promise[Int] task1(1).onComplete(v => v match { case Success(v1) => { val a = task2(v1) val b = task3(v1) a.onComplete(v => v match { case Success(v2) => b.onComplete(v => v match { case Success(v3) => task4(v2 + v3).onComplete(v4 => v4 match { case Success(x) => result.success(x) case Failure(t) => result.failure(t) }) case Failure(t) => result.failure(t) }) case Failure(t) => result.failure(t) }) } case Failure(t) => result.failure(t) }) result.future }
清单 8 看起来很凌乱,幸运的是还有一种简单得多的替代方法:使用 清单 5 中的 flatMap()
代码代替。flatMap()
方法同时处理成功和失败完成类型,无需执行任何更改。
使用 async
最新的 Scala 版本包含在编译期间使用宏 转换代码的能力。目前实现的一个最有用的宏是 async
,它在编译期间将使用 future 的看似顺序的代码转换为异步代码。清单 9 展示了 async
如何简化本教程中使用的任务代码。
清单 9. 结合使用 future 与 async {}
def runAsync(): Future[Int] = { async { val v1 = await(task1(1)) val a = task2(v1) val b = task3(v1) await(task4(await(a) + await(b))) } }
清单 9 中封装的 async {...}
调用了 async
宏。此调用将该代码块声明为异步执行的代码,并在默认情况下异步执行它,然后返回一个 future 表示该代码块的执行结果。在该代码块中,await()
方法(实际上是该宏的一个关键字,而不是一个真正的方法)显示了何处需要一个 future 的结果。async 宏在编译期间修改了 Scala 程序的抽象语法树 (AST),以便将该代码块转换为使用回调的代码,这大体相当于清单 4 的代码。
除了 async {...}
包装器之外,清单 9 中的代码还与 清单 3 中最初的阻塞代码很相似。这主要是这个宏的成就,它抽象化了异步事件的所有复杂性,使它看起来像您在编写简单的线性代码。在幕后,这涉及到大量复杂性。
async
内部原理
如果查看 Scala 编译器从源代码生成的类,就会看到一些具有类似 AsyncHappy$$anonfun$1.class
的名称的内部类。从名称可以猜到,这些类由编译器为异步函数而生成(比如传递给 onSuccess()
或 flatMap()
方法的语句。)
使用 Scala 2.11.1 编译器和 Async 0.9.2 实现,您还会看到一个名为 AsyncUnhappy$stateMachine$macro$1$1.class
的类。这是async
宏生成的实际实现代码,采用状态机的形式来处理异步任务。清单 10 给出了此类的一个部分地方进行了反编译(decompiled)的视图。
清单 10. 反编译后的 AsyncUnhappy$stateMachine$macro$1$1.class
public class AsyncUnhappy$stateMachine$macro$1$1 implements Function1<Try<Object>, BoxedUnit>, Function0.mcV.sp { private int state; private final Promise<Object> result; private int await$macro$3$macro$13; private int await$macro$7$macro$14; private int await$macro$5$macro$15; private int await$macro$11$macro$16; ... public void resume() { ... } public void apply(Try<Object> tr) { int i = this.state; switch (i) { default: throw new MatchError(BoxesRunTime.boxToInteger(i)); case 3: if (tr.isFailure()) { result().complete(tr); } else { this.await$macro$11$macro$16 = BoxesRunTime.unboxToInt(tr.get()); this.state = 4; resume(); } break; case 2: if (tr.isFailure()) { result().complete(tr); } else { this.await$macro$7$macro$14 = BoxesRunTime.unboxToInt(tr.get()); this.state = 3; resume(); } break; case 1: if (tr.isFailure()) { result().complete(tr); } else { this.await$macro$5$macro$15 = BoxesRunTime.unboxToInt(tr.get()); this.state = 2; resume(); } break; case 0: if (tr.isFailure()) { result().complete(tr); } else { this.await$macro$3$macro$13 = BoxesRunTime.unboxToInt(tr.get()); this.state = 1; resume(); } break; } } ... }
清单 10 中的 apply()
方法处理实际的状态更改,估算一个 future 的结果并将输出状态更改为匹配。输入状态会告诉该代码正在估算哪个 future;每个状态值对应于 async 代码块中一个特定的 future。从 清单 10 的部分代码很难了解这一点,但查看其他一些字节码,就可以看到状态代码是与任务匹配的,所以状态 0
表示 task1
的结果符合预期,状态 1
表示 task2
的结果符合预期,依此类推。
resume()
方法并未显示在 清单 10 中,因为反编译器无法确定如何将它转换为 Java 代码。我也不打算探讨这个过程,但通过查看字节码,可以确定 resume()
方法执行了与状态代码上的 Java switch
相似的工作。对于每个非最终状态,resume()
执行适当的代码段来设置下一个预期的 future,最终将 AsyncUnhappy$stateMachine$macro$1$1
实例设置为 future 的 onComplete()
方法的目标。对于最终状态,resume()
将会设置结果值并履行对最终结果的承诺。
您实际上并不需要深入分析生成的代码来理解 async(但它可能很有趣)。关于 async 工作原理的完整描述,请查阅 SIP-22 - Async 提案。
async
限制
由于 async
宏将代码转换为状态机类的方式,该宏的使用有一些限制。最明显的限制是,不能将 await()
嵌套在 async
代码块中的另一个对象或闭包内(包括一个函数定义)。也不能将 await()
嵌套在一个 try
或 catch
内。
除了这些使用限制之外,async
的最大问题是:在调试时,您同样会体验到一些通常与异步代码有关的问题回调,在这种情况下,需要尝试理解没有反映明显的代码结构的调用堆栈。不幸的是,目前的调试器设计无法解决这些问题。这是 Scala 中一个新的工作区域(请参阅 反思调试器。)与此同时,您可以禁用 async
代码块的异步执行,让调试变得更轻松(假设您尝试修复的问题在按顺序执行操作时仍然存在)。
最后,Scala 宏仍是一项我们正在开展的工作。async
有望在未来的版本中成为 Scala 语言的一个正式部分,但只有在 Scala 语言团队对宏的工作方式感到满意时,这种情况才会出现。到那时,无法确保 async
的格式不会发生改变。
http://www.ibm.com/developerworks/cn/java/j-jvmc4/index.html
相关推荐
- **Actor模型**:Scala的Akka库提供了一种处理并发和分布式计算的方式,基于actor模型,简化了并发编程。 4. 兼容性与API更新: Scala 2.11引入了新的API和库,同时保持对2.10.x版本的大部分API兼容。这使得从旧...
- 响应式编程:Scala的Future和Akka框架提供了异步处理和响应式编程的支持。 - 闭包和柯里化:Scala的函数可以捕获和存储外部环境,形成闭包。柯里化允许将多参数函数转换为一系列单参数函数。 5. 类型系统: - ...
10. **未来和承诺**:Scala提供了Future和Promise,用于异步编程,这是处理异步操作和回调的重要工具。 通过深入学习“Scala语言规范.pdf”,你将能全面理解Scala的语法特性、编程理念和最佳实践,从而能够编写出...
8. **XML集成**:Scala直接支持XML语法,可以在Scala程序中直接处理XML数据,增加了处理XML的便利性。 9. **反射**:Scala的反射机制允许程序在运行时检查和修改自身,这对于元编程和动态配置非常有用。 10. **...
- **Future与Promise**:Scala的`scala.concurrent`包提供了Future和Promise机制,用于处理异步操作和并发任务。 #### 四、总结 《Beginning Scala》不仅适合没有编程经验的新手,也适用于那些已经熟悉其他编程语言...
- **Future**:异步编程的工具,用于处理可能需要时间才能完成的操作。 ### 7. Scala与其他技术集成 - **与Java互操作**:Scala可以直接调用Java API,反之亦然,因为它们共享JVM平台。 - **Web开发**:Scala可以...
Scala由Martin Odersky于2003年创建,它的名字是“Scalable Language”的缩写,旨在提供一种可扩展的语言,既能满足大规模并发处理的需求,又能保持代码的简洁性和可读性。Scala运行在Java虚拟机(JVM)上,因此...
- **阻塞与非阻塞**:Scala中的Futures支持非阻塞性编程,但也提供了阻塞性方法来等待异步任务完成。 - **异步库扩展**:除了标准库提供的Futures外,还有许多第三方库扩展了Scala的异步编程能力,如Akka Future等。...
7. ** Futures和Promises**:Scala提供了Futures和Promises,用于异步编程,它们可以帮助处理回调地狱,提高代码的可读性和可维护性。 8. ** Shapeless**:Shapeless是一个元编程库,为处理泛型提供高级工具,如...
6. Scala与Java互操作性:Scala可以在Java项目中无缝集成,因为它是JVM上的语言。可以调用Java库,反之亦然,这使得迁移现有Java项目到Scala变得容易。 7. 函数式编程:Scala支持高阶函数、闭包和不可变数据结构,...
Scala 2.11.4 是 Scala 编程语言的一个版本,主要针对Java虚拟机(JVM)运行。Scala是一种多范式编程语言,融合了面向对象和函数式编程的概念,设计目标是提供一种更高级别的抽象,使代码更加简洁、优雅。在Scala ...
10. **未来和承诺(Future和Promise)**:Scala提供了异步编程的支持,未来(Future)和承诺(Promise)是处理异步操作的重要工具。 11. **Scala REPL**:Scala交互式解释器(Read-Eval-Print Loop)是学习和测试...
1. **高性能与高效率**:Scala 是一种高性能的语言,它被设计成能够高效地运行在 JVM(Java虚拟机)上,这意味着它可以充分利用 Java 生态系统中的强大功能,如垃圾回收机制等。 2. **强大的类型系统**:Scala 拥有...
Actor是异步的、独立运行的实体,通过消息传递进行通信,这有助于简化并发编程的复杂性。 7. **类型系统进阶**:Scala还支持类型参数化、类型成员、类型投影、隐式转换等高级类型系统特性,这为构建复杂、灵活的...
7. **并行和并发**:Scala的并行库利用了Actor模型,这是一种处理并发问题的方法,可以减少线程同步的复杂性。此外,Scala还支持 Futures 和 Promises,用于异步编程。 8. **高阶函数**:Scala中的函数可以作为其他...
通过阅读《Scala程序设计第二版》,开发者不仅可以掌握Scala语言的基础知识,还能了解到如何利用Scala进行高效的并发编程、构建大规模分布式系统,以及如何在实际开发中发挥Scala的灵活性和表达力。对于希望进入...
5. **Future**:在并发编程中,Future表示一个异步计算的结果,提供了非阻塞的等待和结果处理方式。 6. **Implicits**:隐式转换和隐式参数是Scala的一大特色,它们提供了代码的简洁性和灵活性。 **三、使用API文档...
7. **Actor模型**:Scala与Akka框架紧密结合,支持基于Actor的并发编程,Actor是轻量级的并发实体,通过消息传递实现异步通信。 8. **元编程**:Scala的反射和宏(macro)能力允许在编译时进行代码生成和修改,增加...
5. ** Actors模型**:Scala内置了对Akka框架的支持,这是一个用于构建并发和分布式系统的工具,其中Actors模型提供了一种处理异步和并发的高效方法。 6. **集合库**:Scala的集合库是高度优化的,提供了丰富的操作...