我如何用Akka控制流？

Question

class Dispatcher extends Actor {
  def receive = {
    case ResetPassword(to, data) => 
      senderActor ! Mail("resetPassword", to, data)
      Thread.sleep(5000)
    ...
  }
}
class Sender extends Actor {
  def receive = {
    case Mail(to, data) => // send the mail immediately
    ...
  }
}

走这条路对吗？如果没有，流量控制该怎么做？

Answer 1

在不阻塞的情况下做到这一点的一种方法是拥有另一个在任何时候保持发件箱状态的执行元。

class Outbox extends Actor {

  var outbox: List[Mail] = Nil  

  def receive = {
    case a: Mail => {
      outbox = outbox :+ a
    }

    case MailToSend => {
      val maybeMailToSend = outbox.headOption
      outbox = outbox.tail
      senderActor ! maybeMailToSend
    }
  }
}

然后让发件人执行元成为每5秒轮询发件箱的计划执行元。它向发件箱执行元发送askMailToSend消息，如果响应是Some，则发送一个邮件。

它看起来像这样：

system.scheduler.scheduleOnce(5 seconds, senderActor, PollOutbox)

class Sender extends Actor {
  def receive = {
    case PollOutbox => {
      maybeMailToSend = outboxActor ? MailToSend
      ...
    }
  }
}