7

大约两天前，我开始学习 scala actor 框架。为了使我的想法具体化，我决定实现一个基于 TCP 的回显服务器，它可以处理多个同时连接。

这是回显服务器的代码（不包括错误处理）：

class EchoServer extends Actor {
  private var connections = 0

  def act() {
    val serverSocket = new ServerSocket(6789)

    val echoServer = self
    actor { while (true) echoServer ! ("Connected", serverSocket.accept) }

    while (true) {
      receive {
        case ("Connected", connectionSocket: Socket) =>
          connections += 1
          (new ConnectionHandler(this, connectionSocket)).start
        case "Disconnected" =>
          connections -= 1
      }
    }
  }
}

基本上，服务器是一个处理“已连接”和“已断开”消息的 Actor。它将连接委托给匿名参与者，该参与者调用 serverSocket上的accept()方法（阻塞操作）。当连接到达时，它通过“已连接”消息通知服务器，并将套接字传递给它以用于与新连接的客户端通信。ConnectionHandler类的一个实例处理与客户端的实际通信。

这是连接处理程序的代码（包括一些错误处理）：

class ConnectionHandler(server: EchoServer, connectionSocket: Socket)
    extends Actor {

  def act() {
    for (input <- getInputStream; output <- getOutputStream) {
      val handler = self
      actor {
        var continue = true
        while (continue) {
          try {
            val req = input.readLine
            if (req != null) handler ! ("Request", req)
            else continue = false
          } catch {
            case e: IOException => continue = false
          }
        }

        handler ! "Disconnected"
      }

      var connected = true
      while (connected) {
        receive {
          case ("Request", req: String) =>
            try {
              output.writeBytes(req + "\n")
            } catch {
              case e: IOException => connected = false
            }
          case "Disconnected" =>
            connected = false
        }
      }
    }

    close()
    server ! "Disconnected"
  }

  // code for getInputStream(), getOutputStream() and close() methods
}

连接处理程序使用匿名参与者，通过在套接字的输入流上调用readLine()方法（阻塞操作）来等待将请求发送到套接字。当收到请求时，会向处理程序发送“请求”消息，然后处理程序将请求回显给客户端。如果处理程序或匿名参与者遇到了底层套接字的问题，则套接字将关闭，并向回显服务器发送“断开连接”消息，指示客户端已与服务器断开连接。

所以，我可以启动 echo 服务器并让它等待连接。然后我可以打开一个新终端并通过 telnet 连接到服务器。我可以向它发送请求，它会正确响应。现在，如果我打开另一个终端并连接到服务器，服务器会注册连接，但无法为这个新连接启动连接处理程序。当我通过任何现有连接向它发送消息时，我没有立即得到响应。这是有趣的部分。当我终止除一个现有客户端连接之外的所有客户端连接并使客户端 X 保持打开状态时，将返回对我通过客户端 X 发送的请求的所有响应。我做了一些测试并得出结论，即使我调用start() ，后续客户端连接也不会调用act()方法创建连接处理程序的方法。

我想我在连接处理程序中错误地处理了阻塞操作。由于先前的连接是由一个连接处理程序处理的，该连接处理程序阻止了一个匿名参与者等待请求，我认为这个被阻止的参与者正在阻止其他参与者（连接处理程序）启动。

使用 scala 演员时我应该如何处理阻塞操作？

任何帮助将不胜感激。

4

1 回答 1

4

从scala.actors.Actor 的 scaladoc：

注意：在调用 Actor trait 或其伴生对象（例如receive）以外的线程阻塞方法时必须小心。阻塞一个actor内部的底层线程可能会导致其他actor的饥饿。receive这也适用于在调用/之间长时间占用线程的参与者react。

如果参与者使用阻塞操作（例如，阻塞 I/O 的方法），则有几个选项：

  • 可以将运行时系统配置为使用更大的线程池大小（例如，通过设置actors.corePoolSizeJVM 属性）。
  • 可以重写 trait的scheduler方法以返回 a ，它会调整其线程池的大小以避免调用任意阻塞方法的参与者造成饥饿。ActorResizableThreadPoolScheduler
  • The actors.enableForkJoin JVM property can be set to false, in which case a ResizableThreadPoolScheduler is used by default to execute actors.
于 2010-07-27T17:28:55.333 回答