如何在没有 Akka 的情况下实现 Actor 模型？

Question

如何在没有 Akka 的情况下实现简单的 Actor？对于许多（非固定计数）actor 实例、绿色线程、IoC（生命周期、基于 Props 的工厂、ActorRef 的）、监督、背压等，我不需要高性能。只需要顺序性（队列）+ 处理程序 + 状态+ 消息传递。

作为副作用，我实际上需要基于actor的小型管道（带有递归链接）+一些并行actor来优化DSP算法计算。它将在没有传递依赖关系的库中，所以我不希望（并且不能因为它是一个 jar 插件）推动用户创建和传递akkaSystem，该库应该具有尽可能简单和轻量级的界面。我不需要 IoC，因为它只是一个库（一组函数），而不是一个框架——所以它的算法复杂性比结构复杂。但是，我认为 Actor 是描述协议的好工具，实际上我可以将算法分解为少量异步交互的实体，因此它符合我的需求。

为什么不用 Akka

Akka 很重，也就是说：

• 这是一个外部依赖；
• 具有复杂的接口和实现；
• 对于库的用户不透明，例如 - 所有实例都由 akka 的 IoC 管理，因此不能保证一个逻辑参与者始终由同一个实例维护，重新启动将创建一个新实例；
• 需要额外的迁移支持，这与 scala 的迁移支持本身相当。
• 使用jstack/jconsole/jvisualvm 调试akka 的绿色线程也可能更难，因为一个actor 可以作用于任何线程。

当然，Akka 的 jar (1.9Mb) 和内存消耗（每 GB 250 万演员）一点也不重，因此您甚至可以在 Android 上运行它。但众所周知，您应该使用专门的工具来观察和分析用户可能不熟悉的演员（如 Typesafe Activator/Console）（我不会强迫他们学习）。对于企业项目来说一切都很好，因为它几乎总是有 IoC、一些专门的工具和持续迁移，但这对于一个简单的库来说并不是一个好方法。

附：关于依赖。我没有它们，我不想添加任何东西（我什至避免使用 scalaz，它实际上有点适合这里），因为它会导致大量维护 - 我必须保留我的简单库与 Akka 保持同步。

Answer 1

这将使用 FixedThreadPool（以及它的内部任务队列）：

import scala.concurrent._

trait Actor[T] {
  implicit val context = ExecutionContext.fromExecutor(java.util.concurrent.Executors.newFixedThreadPool(1))
  def receive: T => Unit
  def !(m: T) = Future { receive(m) }
}

大小为 1 的 FixedThreadPool 保证了这里的顺序性。当然，如果您需要 100500 个动态创建的 Actor，这不是管理线程的最佳方式，但如果您需要每个应用程序有一些固定数量的 Actor 来实现您的协议，那就没问题了。

用法：

class Ping(pong: => Actor[Int])  extends Actor[Int] {     
      def receive = {
          case m: Int => 
             println(m)
             if (m > 0) pong ! (m - 1)
      }    
}

object System { 
      lazy val ping: Actor[Int] = new Ping(pong) //be careful with lazy vals mutual links between different systems (objects); that's why people prefer ActorRef
      lazy val pong: Actor[Int] = new Ping(ping)
}

System.ping ! 5

结果：

import scala.concurrent._
defined trait Actor
defined class Ping
defined object System
res17: scala.concurrent.Future[Unit] = scala.concurrent.impl.Promise$DefaultPromise@6be61f2c
5
4
3
2
1
0

scala> System.ping ! 5; System.ping ! 7
5
7
4
6
3
5
2
res19: scala.concurrent.Future[Unit] = scala.concurrent.impl.Promise$DefaultPromise@54b053b1
4
1
3
0
2
1
0

此实现使用两个 Java 线程，因此它比没有并行化的计数快“两倍”。

Answer 2

这是 JVM 世界中最小且最高效的 actor，其 API 基于 Viktor Klang 的 Minimalist Scala actor： https://github.com/plokhotnyuk/actors/blob/41eea0277530f86e4f9557b451c7e34345557ce3/src/test/scala/com/github/gist/viktorklang/Actor.scala

它使用起来方便且安全，但在接收消息时类型不安全，并且无法在进程或主机之间发送消息。

主要特点：

• 只有 3 个状态（Stay、Become 和 Die）的最简单的类似 FSM 的 API：https://github.com/plokhotnyuk/actors/blob/41eea0277530f86e4f9557b451c7e34345557ce3/src/test/scala/com/github/gist/viktorklang/Actor.scala#L28-L30

• 极简错误处理 - 正确转发到执行线程的默认异常处理程序：https://github.com/plokhotnyuk/actors/blob/41eea0277530f86e4f9557b451c7e34345557ce3/src/test/scala/com/github/gist/viktorklang/Actor.scala#L52-L53

• 快速异步初始化大约需要 200 ns 才能完成，因此不需要额外的期货/演员来进行耗时的演员初始化：https://github.com/plokhotnyuk/actors/blob/41eea0277530f86e4f9557b451c7e34345557ce3/out0.txt#L447

• 最小的内存占用，即被动状态下约 40 个字节（顺便说一句，new String() 在 JVM 堆中花费相同数量的字节）：https://github.com/plokhotnyuk/actors/blob/41eea0277530f86e4f9557b451c7e34345557ce3/out0.txt#L449

• 消息处理非常高效，4 核 CPU 的吞吐量约为 90M msg/sec：https://github.com/plokhotnyuk/actors/blob/41eea0277530f86e4f9557b451c7e34345557ce3/out0.txt#L466

• 消息发送/接收非常高效，延迟约 100 ns：https://github.com/plokhotnyuk/actors/blob/41eea0277530f86e4f9557b451c7e34345557ce3/out0.txt#L472

• 通过批处理参数调整每个参与者的公平性：https://github.com/plokhotnyuk/actors/blob/41eea0277530f86e4f9557b451c7e34345557ce3/src/test/scala/com/github/gist/viktorklang/Actor.scala#L32

有状态计数器示例：

  def process(self: Address, msg: Any, state: Int): Effect = if (state > 0) { 
     println(msg + " " + state)
     self ! msg
     Become { msg => 
        process(self, msg, state - 1)
     }
  } else Die

  val actor = Actor(self => msg => process(self, msg, 5))

结果：

scala> actor ! "a"
a 5

scala> a 4
a 3
a 2
a 1