Scala + Akka 的奇怪尝试/捕获行为

Question

我正在尝试使用 Akka 为自定义应用程序协议实现 TCP 服务器。我正在尝试遵循此处给出的示例：http://doc.akka.io/docs/akka/2.0/scala/io.html 在 for...yield 循环中执行非阻塞 IO。

我发现当我从 yield 块内抛出异常时，我无法从块外捕获它。我认为我对 Akka 或 Scala 在这里的工作方式存在根本性的误解，如果有任何提示，我将不胜感激。

我将代码归结为：

import akka.actor._
import java.net.InetSocketAddress

class EchoServer(port: Int) extends Actor {

  val state = IO.IterateeRef.Map.async[IO.Handle]()(context.dispatcher)

  override def preStart {
    IOManager(context.system) listen new InetSocketAddress(port)
  }

  def receive = {
    case IO.NewClient(server) =>
      val socket = server.accept()
      state(socket) flatMap (_ => EchoServer.processRequest(socket))
    case IO.Read(socket, bytes) =>
      state(socket)(IO.Chunk(bytes))
    case IO.Closed(socket, cause) =>
      state(socket)(IO.EOF(None))
      state -= socket
  }
}

object EchoServer extends App
{
  def processRequest(socket: IO.SocketHandle): IO.Iteratee[Unit] =
  {
    println( "In process request")
    try {
      for {
        bs <- IO take 1
      } yield {
        println("I'll get here")
        throw new Exception("Hey-o!")
        println("But not here ... as expected")
      }
    } catch {
      case e: Exception => println("And not here ... wtf?"); IO.Done()  // NEVER GETS HERE
    }
  }

  ActorSystem().actorOf(Props(new EchoServer(8080)))
}

也许更方便地遵循这里的要点：https://gist.github.com/2296554

谁能解释为什么在这种情况下控制无法到达我的 catch 块？

我注意到，如果我在 Akka 中打开调试日志记录，我会在输出中看到这条消息：

[DEBUG] [04/03/2012 22:42:25.106] [EchoServerActorSystem-akka.actor.default-dispatcher-1] [Future] Hey-o!

所以我猜异常是由 Akka 调度程序处理的？谁能解释一下这是怎么可能的？

Answer 1

non-blocking IO的要点当然是不能保证何时何地执行。请记住，可以将 for 推导式写为

(IO take 1).map(bs => {
  println("I'll get here"); throw // ...
}

这段代码有什么作用？ IO take 1 返回一些非阻塞的Future 类似的东西，然后通过map 方法附加一个转换函数。 IE。无论何时（无论何时）IO take 1 准备就绪，它都会在结果上应用map。

所有这些都发生在其他线程中（或使用其他实现非阻塞语义的方式），因此try–catch 无法对任何被抛出的Exceptions 做出反应. bs => println(…) … 方法也不会知道您的异常处理。它只知道它应该转换一些输入 bs 并在完成后得到结果。

要吸取的教训：使用非阻塞代码时要避免副作用。如果使用副作用来改变执行流程，尤其如此。

为了实际捕获异常，我认为您必须将其构造如下（未经测试；参见API）：

def processRequest(socket: IO.SocketHandle): IO.Iteratee[Unit] =
  {
    println( "In process request")
    (
      for {
        bs <- IO take 1
      } yield {
        println("I'll get here")
        throw new Exception("Hey-o!")
        println("But not here ... as expected")
      }
    ) recover {
      case e: Exception => println("And here ...?"); IO.Done()
    }
  }