【译】无控制反转的基于事件的程序设计

并发编程是编程界永远绕不开的话题。一是分布式和移动环境天生需要并发&＃xff0c;二是支持并发多线程的多核系统实乃大势所趋。

Actors为并发分布式计算提供了一种适合的计算模型。它使用异步消息通信来实现并发。基于actor的进程模型结合消息的模式匹配&＃xff0c;已在Erlang的成功中被证明是非常高效的。

Erlang是为实时控制系统设计的一种动态类型函数式语言。在电话交换机、网络模拟器&＃xff0c;还有分布式资源控制器中应用广泛。这些系统需要同时激活大量的并发进程&＃xff0c;且因为这些进程处于不断变化中&＃xff0c;想估算它们的内存消耗非常困难。

常规的实现方式——包括操作系统线程以及虚拟机线程——通常过于重量级。主要原因在于&＃xff1a;1&＃xff09;超额配置的栈会快速消耗虚拟地址空间&＃xff1b;2&＃xff09;锁竞争机制通常缺乏合适的管理。因此Erlang在运行时使用的是自建一套并发体系而不是基于底层的操作系统。

Erlang这种轻量级的进程在当前流行的虚拟机上无法适用。而同时标准的虚拟机在Erlang曾大获成功的实时操作领域的地位也越来越重要。

同时智能手机和个人电子设备的普及也带动了其虚拟机的发展。通常情况下这些设备提供的资源都很有限&＃xff0c;如虚拟机可能只能调用几百字节的内存。

这些现状带来重要结果如下&＃xff1a;1&＃xff09;移动设备上的虚拟机提供的服务通常只是桌面、服务器虚拟机的子集。如KVM没有提供自检&＃xff08;introspection&＃xff09;及序列化。2&＃xff09;应用程序所使用的编程抽象必须非常轻量级才能发挥作用。基于线程的并发抽象就显得过于笨重了。除此之外&＃xff0c;编程模型还必须适配虚拟机提供的有限功能。

线程编程常见的代替者是基于事件驱动&＃xff08;Event-based&＃xff09;的编程模型。而直接对显式的事件驱动模型编程是很困难的。

许多编程模型所谓的基于事件编程是通过控制反转来实现的。基于事件的编程意指程序仅仅注册其感兴趣的事件&＃xff0c;从而在特定事件下被唤醒&＃xff08;如&＃xff1a;按下按钮、输入框输入了不同的字符等&＃xff09;&＃xff0c;而非调用阻塞操作&＃xff08;如&＃xff1a;等待获取用户输入&＃xff09;。其逻辑是事件处理器被注册到执行环境并在特定事件发生时被调用。程序自身并不直接调用这些处理器&＃xff0c;而由执行环境来负责向处理器分发事件。因此这些代码的控制逻辑被“反转”了。

上述系统会有两个显而易见的缺点&＃xff1a;1&＃xff09;整个程序的交互逻辑被碎片化到多大事件处理器上。2&＃xff09;处理器之间的控制流在操作共享状态的过程中变得模糊。

为了从轻量级抽象模型中去掉控制反转&＃xff0c;我们试图让actors与线程无关&＃xff08;Thread-less&＃xff09;。我们向‘不友好’的虚拟机中引入了基于事件的actors用作轻量级抽象。‘不友好’指虚拟机不提供对程序执行状态的直接管理。

Actor的中心思想是&＃xff1a;等待中的actor并不是一条阻塞的线程&＃xff0c;而表示为一个捕获其他actor计算结果的闭包。当消息被发送到匹配的actor&＃xff0c;闭包就会立刻执行。闭包的执行发生在发送者的线程上。receive 闭包终止后控制权将回到发送者&＃xff0c;就好像一个过程正常返回一样。如果receive闭包阻塞了随后的接收&＃xff0c;控制权仍将回到发送者&＃xff0c;但同时会抛出一个特殊的异常来释放发送者的调用栈。

这种方案正常运行的必要条件之一是接受者永远不正常返回到它所包含的actor。换句话说&＃xff0c;actor的代码不能依赖于接收方的结束或者阻塞。我们可以在编译时期通过Scala的类型系统来表达这种非返回的属性。不过在实际应用中这并算严格的限制&＃xff0c;因为程序永远可以表达为&＃xff1a;一个actor剩余的计算总是发生在某一个接收下。

据我们所知&＃xff0c;基于事件的actors第一个同时做到了下面两点&＃xff1a;1&＃xff09;在无控制反转时允许交互行为&＃xff1b;2&＃xff09;同时在程序中允许任意的阻塞操作。我们的actor库在消息传递速度及内存消耗上优于其他已有的actor语言几个数量级。我们的实现可以利用多核多进程&＃xff0c;因为互作用可以同时发生在多个处理器中。通过在可移动的运行时系统中扩展这种的基于事件的actors&＃xff0c;我们将展示如何在Scala中实现分布式Erlang的精髓。我们的库几乎已经实现了Erlang所有的原型及内置函数。当前&＃xff0c;我们的运行时系统分别建立在TCP及JXTA点到点框架&＃xff08;Sun微系统对等网络(P2P)标准&＃xff09;这两种标准上&＃xff0c;来保证可移植性。

在具体实现中我们并没有扩展及改变编程语言本身。因此actor库是Scala抽象能力的一个好例子。在2.1.7版本中&＃xff0c;它成为了Scala标准库的一部分&＃xff08;现在已经被akka代替&＃xff09;。

下面和其他相关工作进行比较&＃xff1a;Actalk&＃xff0c;一个Smalltakl-80下的actors库&＃xff0c;扩展了纯Smalltalk对象的最小内核。他们的实现并不是基于事件的&＃xff0c;且Smalltalk-80并不支持多核并发。

Actra扩展了Smalltalk/V虚拟机并使用了提供轻量级进程的基于对象的实时内核。但我们的实现并没有修改虚拟机本身。

Chrysanthakopoulos和Singh讨论过设计和实现基于通道的异步消息库。通道可以被看做特殊的无状态且必须指明接收类型的actors。他们开发了自己的调度程序来替代重量级的操作系统线程来支持连续传递风格&＃xff08;CPS&＃xff09;的代码。使用集群风格的阻塞样式迭代器代码由C#编译器转换而来。

SALSA&＃xff08;简单Actor语言&＃xff0c;系统及架构&＃xff09;扩展了Java的并发并直接支持了actor概念。SALSA程序由预处理器翻译为Java源码以链接一个定制的actor库。对比前面的Smalltalk及通道实现而言&＃xff0c;SALSA在JVM实现actors这点上和我们很相似。本文后面我们将发布两者benchmark的性能测试对比结果。

Timber是一个结合OO和函数式编程的语言&＃xff0c;一般用于实时嵌入式系统。它提供消息传递原型用于交互对象的同步/异步通信。不同于基于事件的actors&＃xff0c;交互对象在长期阻塞时无法调用任何操作。相反它们在计算环境中注册了回调方法以代替它们自身执行操作。

Frugal objects(FROBs) 是通过类型化事件来通信的分布式交互对象。本质上说FROBs是基于事件的计算模型&＃xff0c;就像我们的actors一样。但是FROBs和基于事件的actors两者的目标是正交的。前者提供了适用于资源受限设备上的计算模型。后者目的是为基于事件的actors提供一个编程模型&＃xff08;如&＃xff1a;方便的语法等&＃xff09;&＃xff0c;当然也包括FROBs。目前为止&＃xff0c;FROBs只能使用Java底层API。未来我们计划双方合作集成目标。

本文剩余章节结构如下。第二章展示了如何用Scala库来表示方便的、基于线程的actors。第三章介绍如何修改上节actor模型&＃xff0c;使其变为基于消息。第四章简述Scala分布式actor。第五章评估了我们actor库的性能。第六章总结。

本章描述如何用Scala库实现类似Erlang的抽象。Actors是异步消息传递的、自包含的有效逻辑实体。下面将展示了一个计数器actor。此actor重复一个receive操作&＃xff0c;等待三种消息&＃xff1a;

• Incr消息&＃xff0c;计数器值&＃43;1
• Value消息&＃xff0c;计数器将当前值发给指定的actor
• Lock消息让事情更加有趣。当收到一个Lock&＃xff0c;计数器会将当前值传递给指定的actor&＃xff0c;然后阻塞直到收到一个UnLock消息。此消息会指定一个计数器开始继续的值。因此除非它解锁&＃xff0c;其他进程将无法观测到一个锁定的计数器状态。

def loop(value: Int): Unit &＃61; {Console.println("Value: " &＃43; value)receive {case Incr() &＃61;> loop(value &＃43; 1)case Value(a) &＃61;> a ! value; loop(value)case Lock(a) &＃61;> a ! valuereceive { case UnLock(v) &＃61;> loop(v) }case _ &＃61;> loop(value)}}

除了上面三种消息&＃xff0c;其它的都会被直接丢弃。和计数器典型通信方式如下&＃xff1a;

val counter &＃61; new Countercounter.start()counter ! Incr()counter ! Value(this)receive { case cvalue &＃61;> Console.println(cvalue) }

首先新建一个计数器&＃xff0c;启动&＃xff0c;发送Incr()消息来使其自增&＃xff0c;然后发送Value来查询当前值。然后等待计数器actor的响应。一旦收到响应就打印它的值&＃xff08;值应该是1&＃xff0c;除非有其他actor同时使用了此计数器&＃xff09;。

此模型中的消息是任意对象。在通道编程中一个通道必须指定自身能够处理的消息类型&＃xff0c;而一个actor可以收到任何消息类型。

我们的例子中&＃xff0c;actors之间使用下面四种消息类来通信&＃xff1a;

case class Incr()case class Value(a: Actor)case class Lock(a: Actor)case class UnLock(value: Int)

所有类都用case修饰符来使用构造器模式。所有类都仅有声明而没有主体。Incr类构造器没有参数&＃xff0c;Value和Lock类构造器都有一个Actor参数&＃xff0c;UnLock有一个整形构造器参数。

a!m 表示将消息m发给actor a。这个通信是异步的&＃xff0c;即如果a没有准备好接收m&＃xff0c;m将在a的等待队列中排队并立即返回结束操作。

消息在接收结构中处理&＃xff1a;

receive {case p1 &＃61;> e1// ...case pn &＃61;> en}

这里actor等待队列中的消息会和p1到pn的模式依次进行匹配。模式由构造器和变量组成。每个构造器命名了一个用例类&＃xff08;case class&＃xff09;用来匹配本类的所有实例。每个变量模式匹配每个绑定的值。如模式Value(a)匹配了Value类的所有实例v&＃xff0c;并且将变量a绑定到v的构造函数参数中。

接受时将选择actor等待队列中第一条消息来进行匹配操作。如果匹配到了模式pi&＃xff0c;相应的ei将会被执行。如果没有消息匹配任意一个模式&＃xff0c;actor将会被挂起等待新的消息到达。

在上面例子显示Scala语言简直是为actor并发量身定制的。而事实上并非如此。Scala仅假设了底层主机的基础线程模型。所有高级操作都被定义为Scala库的类和方法。本章剩余部分&＃xff0c;我们来看看幕后每个构造器是如何定义及实现的。

一个actor只是定义了&＃xff01;和 receive 方法用来接收发送消息的主机环境线程类的子类。

abstract class Actor extends Thread {private var mailbox: List[Any]def !(msg: Any) &＃61; ...def receive[a](f: PartialFunction[Any, a]): a &＃61;...// ...
}

!方法用来向actor发送消息。a!m是a.!(m)的缩写&＃xff0c;如x&＃43;y是x.&＃43;(y)的缩写一样。此方法有两个作用&＃xff0c;一是将消息发到actor等待队列mailbox——List[Any]类型的私有字段。第二如果能处理此发送消息的接收actor当前处于挂起状态&＃xff0c;那么它将被恢复执行。

receive{...} 结构就有趣的多。Scala将括号模式匹配表达式视为头等对象并作为参数传递给接收方。参数类型是一元函数Function1子类PartialFunction的实例。它们的定义如下&＃xff1a;

abstract class Function1[-A,&＃43;B] {def apply(x: A): B
}abstract class PartialFunction[-A,&＃43;B] extends Function1[A,B] {def isDefinedAt(x: A): Boolean
}

我们可以看到函数是有着apply方法的对象。偏函数(Partial function)则多了一个isDefinedAt方法&＃xff0c;用来识别指定的a是否有此函数。这两个类都是泛型&＃xff0c;第一个类型参数指明函数的参数类型&＃xff0c;第二个类型参数指明了返回值的类型。
{case p1 &＃61;> e1; ...; case pn &＃61;> en} 也是一种偏函数&＃xff0c;其方法如下定义&＃xff1a;

• isDefinedAt&＃xff1a;pi匹配成功返回true&＃xff0c;匹配失败返回false
• apply&＃xff1a;第一个成功匹配的pi返回ei。如果匹配不到任何模式&＃xff0c;将抛出MatchError异常将被抛出。

receive首先按序扫描等待队列中的消息。如果和接收方参数 f匹配成功&＃xff0c;那么相应的消息将被从队列中移除并用于f。反之&＃xff0c;如果队列中的所有消息对于f.isDefinedAt(m)方法都返回false&＃xff0c;此actor对应的线程将被挂起。

上面总结了基于线程的actors。当然并没有涵盖Scala的actor库的全部功能。例如&＃xff0c;receiveWithin方法可以指定actor接收消息时的等待超时时间。如果超时一个特殊的模式 TIMEOUT() 将会被触发。超时可用来挂起actor&＃xff0c;刷新全部mailbox或用来实现优先级消息。

基于线程的actors作为线程的高级别抽象很有用&＃xff0c;因为它们去掉了异步消息传递时易于出错的共享锁。但是和线程一样&＃xff0c;它们给JVM之类的平台带来了性能损失&＃xff0c;如限制了其可扩展性。下一章我们将展示如何从actor模型中去掉线程。

从逻辑上来说 actor 的执行并没有绑定到某个线程上。但是&＃xff0c;所有的actor模型实现都将单个的actor关联到独立的线程、甚至是操作系统进程上。

在Scala中&＃xff0c;标准库的线程抽象映射到相应目标平台的线程实现上&＃xff0c;如当前的JVM和微软的CLR。

为了克服伸缩性问题&＃xff0c;我们提出了一个基于事件的actor实现&＃xff1a;1&＃xff09;线程无关&＃xff1b;2&＃xff09;两个事件间的运算被认为已完成。一个事件表示一个新消息到达actor的队列。

执行例子

首先&＃xff0c;我们直观的介绍此实现是如何工作的。让我们回到第二章的计数器例子。

首先创建一个可锁定的计数器实例c&＃xff0c;c有着一个空队列。启动后&＃xff0c;c立即阻塞并等待消息匹配。另一个actor p向c发送一个Lock(p)的消息(c ! Lock(p))。Lock消息的到达使得c从阻塞中恢复并获得控制权。c恢复自己的阻塞receive语句。这里我们重用了发送方的线程&＃xff0c;而不是在接收actor自己的线程上执行。

根据receive的语义&＃xff0c;新的消息匹配上之后被选中并从队列中移除。随后将以匹配的变量执行相应action&＃xff1a;

{ case Incr() &＃61;> loop(value &＃43; 1)case Value(a) &＃61;> a ! value; loop(value)case Lock(a) &＃61;> a ! valuereceive { case UnLock(v) &＃61;> loop(v) }case _ &＃61;> loop(value)
}.apply(Lock(p))

可以直观的缩写如下&＃xff1a;

p!value
receive {case UnLock(v) &＃61;> loop(v) }

 执行之后&＃xff0c;由于c的队列中没有其他消息&＃xff0c;receive再次被阻塞。请记住此时我们仍然在p的原始消息发送中&＃xff08;发送还没有返回&＃xff09;。因此&＃xff0c;阻塞当前线程&＃xff08;通过调用wait()方法&＃xff09;将同样会阻塞p。

这并不合理&＃xff0c;因为在我们的编程模型中发送操作 ! 有着非阻塞的语义。相反&＃xff0c;我们希望的是挂起 c 的同时 p 仍然继续运行。为了做到这一点&＃xff0c;我们在阻塞的receive记住c剩余的计算。本例中&＃xff0c;它以如下闭包的形式保存&＃xff1a;

receive { case UnLock(v) &＃61;> loop(v) }

第二&＃xff0c;让 p 对于发送的操作返回&＃xff0c;我们需要将运行时堆栈解除到控制转移到c的地方。为了做到这一点我们抛出了一个特殊的异常。 ! 方法会捕获此异常&＃xff0c;并正常返回&＃xff0c;保持其非阻塞语义。

一般来说&＃xff0c;保存一个闭包来捕获actor剩余的计算是不够的。考虑一个actor执行如下语句的场景&＃xff1a;

val x &＃61; receive { case y &＃61;> f(y) }
g(x)

这里&＃xff0c;receive操作产生一个值并传递给一个函数。假设 receive 阻塞了。我们需要在这个阻塞的 receive 操作中保存剩余的计算。

为了保存receive余下的语句信息&＃xff0c;我们需要保存调用栈&＃xff0c;或者捕获&＃xff08;最高级别的&＃xff09;continuation。由于安全性方面的原因&＃xff0c;类似JVM这样的虚拟机并没有提供显式的栈管理。因此&＃xff0c;语言本身要实现最高级别的continuation 就必须在堆上模拟运行时栈空间&＃xff0c;这会带来一系列性能上的问题。此外&＃xff0c;这样做会导致原生VM上的程序调试工具无法找到运行时信息&＃xff0c;因为实际上它们被放到了堆上。现有的工具将无法使用。

因此&＃xff0c;大多自身支持continuation的语言&＃xff08;Scheme、Ruby&＃xff09;在一些不支持此特性的虚拟机&＃xff08;JRuby&＃xff09;上都选择了不支持最高级别continuation端口。Scala同样也不支持最高级别continuation&＃xff0c;主要是因为和现有Java代码的兼容及互操作性。

总结一下&＃xff0c;管理调用receive之后的操作信息需要修改编译器或者虚拟机。但是我们期望提供一个基于代码库的方式来避免这些修改。

相对的&＃xff0c;我们需要receive方法永远不正常返回。因此处理正常返回的部分可以被省略掉。进一步&＃xff0c;我们可以通过Scala类型系统在编译期保证这一点&＃xff0c;即返回类型为Nothing之后的代码不执行。注意&＃xff0c;抛出异常这种返回仍然是可以的。事实上&＃xff0c;上面已经提到过我们的receive正是依赖这一点。

使用一个不返回的receive&＃xff0c;上面的例子会看起来如下&＃xff1a;

receive{ case y &＃61;> x &＃61; f(y); g(x) }

本质上&＃xff0c;actor剩下的计算将在各个匹配分支中参数的receive函数中被调用&＃xff08;continuation passing风格&＃xff09;。

单线程Actor

我们期望避免控制反转的receive在sender上执行。如果所有的actor都在同一个线程上运行&＃xff0c;向actor A发送一个消息将唤醒执行导致A挂起的receive。下面的代码展示了单线程actors发送操作的一个简单实现&＃xff1a;

def !(msg: Any): Unit &＃61; {mailbox &＃43;&＃61; msgif (continuation !&＃61; null && continuation.isDefinedAt(msg))try {receive(continuation)}catch {case Done &＃61;> // do nothing}
}

要发送的消息被追加到发送目标actor队列的末尾。我们用A来表示目标actor。如果continuation属性被设置为非空&＃xff0c;那么A将会挂起并等待一个合适的消息&＃xff08;否则A不执行receive调用&＃xff09;。这里continuation指的是最后一个被调用的阻塞receive的偏函数&＃xff08;闭包&＃xff09;&＃xff0c;我们可以测试新的消息是否可以让A继续。

注意&＃xff0c;如果我们将一个阻塞的receive(f)保存为continuation&＃xff0c;我们将无法进行判断而只能执行此continuation。如果新的消息无法匹配任意模式&＃xff0c;receive会重复遍历队列中的所有消息试图找到匹配的消息。当然&＃xff0c;这种尝试是徒劳的&＃xff0c;因为只有新添加的消息才能使A继续下去。

如果A能够处理新来的消息&＃xff0c;我们让A继续直到它阻塞于一个嵌套的receive(g)&＃xff0c;或者直到它结束自己的计算。对前一种情形&＃xff0c;我们保存g的闭包作为A的continuation。然后由于其非阻塞语义&＃xff0c;发起A执行的发送操作将返回。因此阻塞的receive将抛出一个类型为Done的特殊异常&＃xff0c;此异常将在发送操作&＃xff08;!&＃xff09;中被捕获。此技巧可以释放堆栈到控制权被转给A的地方。来为了解释发送操作是如何运作的&＃xff0c;我们深入到reveive实现中。

receive方法从actor的队列中选择消息&＃xff0c;然后负责在不解析上下文的同时保存continuation&＃xff1a;

def receive(f: PartialFunction[Any, unit]): Nothing &＃61; {mailbox.dequeueFirst(f.isDefinedAt) match {case Some(msg) &＃61;> continuation &＃61; nullf(msg)case None &＃61;> continuation &＃61; f}throw new Done
}

我们从队列中取出第一个可以匹配某个偏函数定义的消息&＃xff0c;此偏函数作为receive的参数被提供。注意 f.isDefinedAt 的类型为 Any&＃61;>boolean。结果的对象类型为Option[Any]&＃xff0c;它的两种情况可以用两种模式来分别匹配。当一个消息出队时&＃xff0c;我们首先重置当前被保存的continuation。这将避免因为调用的 f(msg) 中有向当前actor的发送请求而导致之前的continuation被多次调用。

如果队列中找不到匹配的消息&＃xff0c;我们记住作为f闭包的continuation。无论哪种情形我们都需要通过抛出Done来禁止解析上下文&＃xff0c;此时请求发送方正常地继续执行。

多线程Actors

为了利用多线程处理器我们希望在多个线程上并发计算。依赖于现代的虚拟机在多个处理器上可以同时执行多个虚拟机线程。

调度器决定一个指定负载的actors需要多少线程&＃xff0c;并自然地实现特定的调度策略。因为其异步性&＃xff0c;发送消息将引入一个并发活动&＃xff0c;即恢复先前被挂起的actor。我们将该活动封装在一个任务中并交给调度器&＃xff08;从某种意义上说这是一次重发&＃xff09;&＃xff1a;

def send(msg: Any): Unit &＃61; synchronized {if(continuation !&＃61; null&& continuation.isDefinedAt(msg)&& !scheduled) {scheduled &＃61; trueScheduler.putTask(new ReceiverTask(this, msg))} else mailbox &＃43;&＃61; msg
}

如果在发送时发现接收actor A的continuation未定义&＃xff0c;A将不等待消息。通常这是因为A的任务当前已预订但是未执行。除非接收方actor正处于等待中、并且能够处理此消息&＃xff0c;否则发送操作都会将消息追加到队列中。此时我们向调度器提交新的任务来安排执行接收方actor continuation。

调度器维持了一个工作线程池来执行任务。一个ReceiverTask就是接收到一个特殊消息的java.lang.Runnable实例&＃xff0c;且有一个异常处理器来禁止上下文的解析&＃xff1a;

class RecervierTask(actor: Actor, msg: Any) extends Runnable {def run(): unit &＃61; {try {actor receiveMsg msg}catch {case Done &＃61;> // do nothing }}
}

receiveMsg是receive的特殊形式&＃xff0c;用actor的continuation来处理指定消息。

Actors无法阻止永久的阻塞操作。接下来我们描述一个在阻塞情况下也能正常进行的调度器。

阻塞操作

我们实现基于事件特性源于下面两个事实&＃xff1a;1&＃xff09;actors是线程无关的&＃xff1b;2&＃xff09;在各消息的到达之间&＃xff0c;运算被认为已完成。第二点在事件驱动的系统中很常见&＃xff0c;它反映了我们对于大多数actor的角色交互的假设。与通信开销相比&＃xff0c;各条消息本身需要的计算通常要短得多。

虽然如此&＃xff0c;我们也希望能支持长期运行的、CPU绑定的actors。这种actors将不阻止其它actors运行。如果某个阻塞actor妨碍到其它actor甚至引起整个程序无响应将是很不幸的。

在用户级的线程包我们遇到了同样的问题&＃xff1a;进程在某些特定的程序节点将控制权交给调度器。在这些节点之间&＃xff0c;无法阻止调用阻塞操作或执行无限循环。比如一个actor可能调用一个会引起系统阻塞的本地方法。

对我们而言&＃xff0c;调度器仅仅在发送的消息导致另一个actor恢复时才执行。因为发送操作不允许阻塞&＃xff0c;所以&＃xff08;被恢复的&＃xff09;接收方需要在其他线程上执行。这样即使接收方阻塞了也不会影响到发送方。

由于调度器可能没有任何空闲的工作线程&＃xff08;即全部处于阻塞状态&＃xff09;&＃xff0c;因此它需要按需创建新线程。但如果有至少一个可用线程&＃xff08;该线程正在执行某个actor&＃xff09;&＃xff0c;我们都不会创建新线程。这样做是为了防止在没有阻塞操作时创建过多的线程。

Actors仍然是线程无关的&＃xff0c;即使&＃xff1a;每次actor由于一个阻塞的&＃xff08;不成功的&＃xff09;receive而挂起时&＃xff0c;它将从自己的线程上分离而非去阻塞它。当执行完了一个接受方任务&＃xff0c;线程重新变为空闲状态。然后会向调度器请求新任务。因此线程在执行多个actors时重用了。

通过这种方式&＃xff0c;基于actor的应用在低并发时甚至可以只需要两个线程就能执行&＃xff0c;而不管同一时间活跃的actor数量。

不幸的是&＃xff0c;用户级的代码无法得知JVM中的一条线程是否处于阻塞态。因此我们实现了简单的模拟来估计执行actor的线程是否阻塞了。

基本的想法是actor执行时向调度器提供心跳。即send&＃xff08;!&＃xff09;和receive方法调用调度器的tick方法。调度器检查当前执行actor的线程&＃xff0c;然后更新其时间戳。当有新的任务被提交到调度器&＃xff0c;调度器首先检查是否所有线程都处于阻塞态。有着‘近期’时间戳的线程被认为是非阻塞的。只有所有工作线程都被判定为阻塞时&＃xff0c;才会创建新的工作线程。否则任务将被简单的追加到队列中等待工作线程来消费。下面展示了调度器实现的主要部分。

注意到其中使用了近似这个词&＃xff0c;因为想明确区分开阻塞的和那些仅仅是执行长时间任务的线程是不可能的。这就意味着计算绑定的actors在它们自己的线程上执行。

def execute(item: ReceiverTask): Unit &＃61; synchronized {if (idle.length > 0) {val worker &＃61; idle.dequeueexcuting.update(item.actor, worker)worker.execute(item)} else {val iter &＃61; workers.elementsvar foundBusy &＃61; falsewhile (iter.hasNest && !foundBusy) {val worker &＃61; iter.nextticks.get(worker) match {case None &＃61;> foundBusy &＃61; truecase Some(ts) &＃61;> val currTime &＃61; System.currentTimeMillisif (currTime - ts

对于某些应用来说&＃xff0c;有必要根据运行时配置来优化空闲线程的数量。使用我们的库很容易实现这类自定义调度器。