仅使用Akka actors作为邮箱的死锁

Question

我只想使用Akka演员作为邮箱，即我想创建n个线程，每个线程创建1个远程演员。

每个线程都获得对其他线程的所有远程参与者的引用，这样它们就可以通过各自的参与者向彼此发送消息。

参与者定义如下：

case class Receive
case class GroupReceive(id: Symbol)
case class GroupMsg[T](id: Symbol, msg: T)
class FooParActor(val distributor: Distributor) extends Actor
  with Stash {
  import context._

  val globalRank: Int = distributor.globalRank

  def doReceive(realSender: ActorRef, ID: Symbol) {
    unstashAll()
    become({
      case GroupMsg(ID, msg) =>
        realSender ! msg
        unbecome()
      case GroupMsg(otherId, msg) =>
        println(globalRank + ": stashing " + otherId)
        unbecome()
      case x => sys.error("bad msg: " + x)
    }, discardOld = false)
  }

  def receive = {
    case GroupReceive(id) =>
      doReceive(sender, id)
    case GroupMsg(id, x) =>
      stash()
    case x => sys.error("bad msg: " + x)
  }

}

def groupRcv[T](id:Symbol) = Await.result(aref ? GroupReceive(id), timeout.duration).asInstanceOf[T]

其中aref是对该线程的本地参与者的引用。

我有时会遇到上述模式的死锁（超时5秒），即使使用非常简单的方法和很小的消息也是如此。我将问题缩小到actors在收到groupreceive(id)消息后，但在输入doReceive（...）的第一个大小写之前停顿:case GroupMsg(id,msg)=>。

我做了打印输出跟踪，以检查演员在执行doReceive调用之前是否确实有消息，似乎出于某种原因，他们只是不处理这些消息。上面给出的代码可以进入groupmsg()从fooparactor的stash中丢失的状态吗？或者，在接收到groupreceive()消息后，参与者是否有其他方法进入死锁？

Answer 1

您正在使用await.result()，但没有共享您在哪里执行：如果您在参与者应该运行的线程上调用grouprcv，那么您当然会出现饥饿（即目标参与者没有可运行的线程，因此它永远不会完成请求）。

您似乎以一种不健康的方式将基于线程的并发与actors混合在一起，但由于您只是暗示它而不显示代码，所以我只能给您一个宽泛的建议，不要这样做。当编程演员时，忘记线程；那些是由Akka管理的。特别是不要滥用Akka的线程（例如，wauit.result可能在您自己的外部线程池中工作，尽管几乎总有更好的替代方案）。

最后，如果您使用actors只是为了创建“一个带有邮箱的线程”，那么Akka就帮不了您，您将遇到所有常见的传统并发陷阱。