使用Akka Actors处理多个TCP连接

Question

我正在尝试使用akka actor 设置一个简单的TCP服务器,这应该允许多个客户端同时连接.我将问题简化为以下简单程序:

package actorfail
import akka.actor._, akka.io._, akka.util._
import scala.collection.mutable._
import java.net._

case class Foo()

class ConnHandler(conn: ActorRef) extends Actor {
  def receive = {
    case Foo() => conn ! Tcp.Write(ByteString("foo\n"))
  }
}

class Server(conns: ArrayBuffer[ActorRef]) extends Actor {
  import context.system
  println("Listing on 127.0.0.1:9191")
  IO(Tcp) ! Tcp.Bind(self, new InetSocketAddress("127.0.0.1", 9191))
  def receive = {
    case Tcp.Connected(remote, local) =>
      val handler = context.actorOf(Props(new ConnHandler(sender)))
      sender ! Tcp.Register(handler)
      conns.append(handler)
  }
}

object Main {
  def main(args: Array[String]) {
    implicit val system = ActorSystem("Test")
    val conns = new ArrayBuffer[ActorRef]()
    val server = system.actorOf(Props(new Server(conns)))
    while (true)  {
      println(s"Sending some foos")
      for (c <- conns) c ! Foo()
      Thread.sleep(1000)
    }
  }
}

它绑定到localhost:9191并接受多个连接,将连接处理程序添加到全局数组并定期将字符串发送"foo"到每个连接.现在,当我尝试同时连接多个客户端时,只有第一个客户端获得"foo".当我打开第二个连接时,它不会被发送任何foo,而是我得到以下类型的日志消息:

Sending some foos
[INFO] [03/27/2015 21:24:07.331] [Test-akka.actor.default-dispatcher-6] [akka://Test/deadLetters] Message [akka.io.Tcp$Write] from Actor[akka://Test/user/$a/$b#-308726290] to Actor[akka://Test/deadLetters] was not delivered. [7] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

我理解这意味着我们尝试发送Tcp.Write命令的目标actor 不再接受消息.但那是为什么呢？你能帮我理解潜在的问题吗？我怎样才能做到这一点？

Answer 1

上面的代码有两个问题:

• 在actor消息中发送可变状态并以非线程安全方式对其进行变更
• 包括道具中的不稳定引用

在我详细说明之前,请考虑阅读文档,这里和这里,这些都包含在内.

可变消息

ArrayBuffer不是线程安全的,但是你将它从主例程传递给不同的actor,然后他们独立地(并发地)修改它.这将导致更新丢失或数据结构本身损坏.另一方面,如果没有适当的同步,则不能保证主线程将看到修改,因为编译器原则上可以确定缓冲区在while循环内不会改变并相应地优化代码.

演员只会发送消息,而不是依赖共享的可变状态.在这种情况下,解决方案是将while循环提升为一个actor(但是self在一秒钟后调度消息而不是阻塞Thread.sleep(1000)调用).然后只需要ActorRef为这个foo发送方actor 传递连接处理程序,它们将向它发送一条消息来注册它们,然后该actor将活动连接列表保存在它封装的范围内.这样做的好处是,您可以使用DeathWatch在终止时删除连接.

道具中不稳定的参考文献

有问题的代码: Props(new ConnHandler(sender))

道具是从一个演员工厂构建的,在这种情况下,它被视为一个名字参数; new每当这样的actor被初始化时,可能在不同的线程上,在稍后的时间评估整个表达式.这意味着sender稍后也会从执行的上下文中进行评估,因此很可能deadLetters(如果父actor当前没有运行 - 如果是,sender则可能完全指向错误的actor).

这里的解决方案记录在这里.