如何使用Akka控制流量？

Question

我已经读到使用Akka时的一个重要规则是避免任何阻塞输入/输出操作,轮询,忙等待,睡眠等等.但是,如果我真的需要一些流量控制呢？

我正在使用Akka演员向我们的客户发送邮件,并且对邮件服务器友好,每5秒发送一封邮件.我的计划是使用调度程序actor执行流控制,使用发送方actor执行邮件发送工作.

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

在没有阻塞的情况下执行此操作的一种方法是让另一个actor随时保持发件箱状态.

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
    }
  }
}

然后让Sender演员成为预定演员,每隔5秒轮询一次发件箱.它向actor 发送一条ask MailToSend消息,Outbox如果响应是Some,则发出一封邮件.

它看起来像这样:

调度发件人actor:

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

发件人演员:

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