73

我目前正在尝试开始使用 Akka，但遇到了一个奇怪的问题。我的演员有以下代码：

class AkkaWorkerFT extends Actor {
  def receive = {
    case Work(n, c) if n < 0 => throw new Exception("Negative number")
    case Work(n, c) => self reply n.isProbablePrime(c);
  }
}

这就是我开始我的工人的方式：

val workers = Vector.fill(nrOfWorkers)(actorOf[AkkaWorkerFT].start());
val router = Routing.loadBalancerActor(SmallestMailboxFirstIterator(workers)).start()

这就是我关闭一切的方式：

futures.foreach( _.await )
router ! Broadcast(PoisonPill)
router ! PoisonPill

现在发生的情况是，如果我向工作人员发送 n > 0 的消息（不抛出异常），一切正常并且应用程序正常关闭。但是，只要我向它发送一条导致异常的消息，应用程序就不会终止，因为仍然有一个演员在运行，但我不知道它来自哪里。

如果有帮助，这是相关线程的堆栈：

  Thread [akka:event-driven:dispatcher:event:handler-6] (Suspended) 
    Unsafe.park(boolean, long) line: not available [native method]  
    LockSupport.park(Object) line: 158  
    AbstractQueuedSynchronizer$ConditionObject.await() line: 1987   
    LinkedBlockingQueue<E>.take() line: 399 
    ThreadPoolExecutor.getTask() line: 947  
    ThreadPoolExecutor$Worker.run() line: 907   
    MonitorableThread(Thread).run() line: 680   
    MonitorableThread.run() line: 182

PS：没有终止的线程不是任何工作线程，因为我添加了一个 postStop 回调，它们每个都正确停止。

PPS：Actors.registry.shutdownAll解决问题，但我认为shutdownAll应该只作为最后的手段，不是吗？

4

3 回答 3

23

处理 akka actor 内部问题的正确方法不是抛出异常，而是设置主管层次结构

“在并发代码中抛出异常（假设我们使用的是非链接的 Actor），只会炸毁当前执行 Actor 的线程。

没有办法发现事情出错了（除了检查堆栈跟踪）。你对此无能为力。”

请参阅通过主管层次结构实现容错 (1.2)

* 注意 *以上对于旧版本的 Akka (1.2) 是正确的。在较新的版本（例如 2.2）中，您仍然需要设置主管层次结构，但它会捕获子进程抛出的异常。例如

class Child extends Actor {
    var state = 0
    def receive = {
      case ex: Exception ⇒ throw ex
      case x: Int        ⇒ state = x
      case "get"         ⇒ sender ! state
    }
  }

在主管中：

class Supervisor extends Actor {
    import akka.actor.OneForOneStrategy
    import akka.actor.SupervisorStrategy._
    import scala.concurrent.duration._

    override val supervisorStrategy =
      OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
        case _: ArithmeticException      ⇒ Resume
        case _: NullPointerException     ⇒ Restart
        case _: IllegalArgumentException ⇒ Stop
        case _: Exception                ⇒ Escalate
      }

    def receive = {
      case p: Props ⇒ sender ! context.actorOf(p)
    }
  }

请参阅通过主管层次结构实现容错 (2.2)

于 2011-11-11T18:34:30.417 回答
2

正如 Viktor 建议的那样，关闭日志记录以确保事情终止，这有点奇怪。你可以做的是：

EventHandler.shutdown()

干净地关闭所有（记录器）侦听器，这些侦听器在异常发生后保持世界运行：

def shutdown() {
  foreachListener(_.stop())
  EventHandlerDispatcher.shutdown()
}
于 2012-01-05T19:36:25.067 回答
-3

打开记录仪akka.conf

于 2011-11-09T09:17:11.557 回答