首页 文章

Akka演员，期货和关闭

提问于
浏览 次
12

我在Akka docs中读到，关闭封闭演员的变量是危险的 .

警告在这种情况下，您需要小心避免关闭包含actor的引用，即不要在匿名Actor类中调用封闭actor上的方法 . 这会打破actor封装，并可能引入同步错误和竞争条件，因为其他actor的代码将同时调度到封闭的actor .

现在，我有两个演员，其中一个从第二个请求某些东西，并对结果做了一些事情 . 在下面的这个例子中，我放在一起，演员Accumulator从演员NumberGenerator中检索数字并将它们相加，并沿途报告总和 .

这可以通过至少两种不同的方式完成，如本例所示，具有两种不同的接收功能（A对B） . 两者之间的区别在于A不会关闭计数器变量;相反，它等待一个整数并将其求和，而B创建一个关闭计数器并完成总和的Future . 如果我正确理解这是如何工作的，那么这是在为处理onSuccess而创建的匿名actor中发生的 .

import com.esotericsoftware.minlog.Log

import akka.actor.{Actor, Props}
import akka.pattern.{ask, pipe}
import akka.util.Timeout
import akka.util.duration._

case object Start
case object Request


object ActorTest {
  var wake = 0

  val accRef = Main.actorSystem.actorOf(Props[Accumulator], name = "accumulator")
  val genRef = Main.actorSystem.actorOf(Props[NumberGenerator], name = "generator")

  Log.info("ActorTest", "Starting !")

  accRef ! Start
}

class Accumulator extends Actor {
  var counter = 0

  implicit val timeout = Timeout(5 seconds)

  // A: WITHOUT CLOSURE
  def receive = {
    case Start => ask(ActorTest.genRef, Request).mapTo[Int] pipeTo self
    case x: Int => counter += x; Log.info("Accumulator", "counter = " + counter); self ! Start
  }
  // B: WITH CLOSURE
  def receive = {
    case Start => ask(ActorTest.genRef, Request).mapTo[Int] onSuccess {
      case x: Int => counter += x; Log.info("Accumulator", "counter = " + counter); self ! Start
    }
  }
}

class NumberGenerator extends Actor {
  val rand = new java.util.Random()

  def receive = {
    case Request => sender ! rand.nextInt(11)-5
  }
}

在这种情况下使用闭包绝对是邪恶的吗？当然我可以使用AtomicInteger而不是Int，或者在某些网络场景中使用netty，在threadsafe Channels 上发出写操作，但这不是我的观点 .

提出要求荒谬的风险：有没有办法让Future的onSuccess在 this 中执行而不是匿名的中间行为者， without 在接收函数中定义一个案例？

EDIT

更明确地说，我的问题是：有没有办法迫使一系列期货在与给定的Actor相同的线程中运行？

scala closures actor akka future

2 回答

  • 5

    问题是 onSuccess 将在与演员的 receive 将要运行的线程不同的线程中运行 . 您可以使用 pipeTo 方法，或使用Agent . 使 counter AtomicInteger 可以解决问题，但它不是那么干净 - 也就是说，它打破了Actor模型 .

    回复于
  • 5

    实现此类设计的最简单方法是使用“即发即忘”语义：

    class Accumulator extends Actor {
  private[this] var counter = 0

  def receive = {
    case Start => ActorTest.genRef ! Request
    case x: Int => {
      counter += x
      Log.info("Accumulator", "counter = " + counter)
      self ! Start
    }
  }
}

    此解决方案完全异步，您不需要任何超时 .

    回复于

相关问题