如何使用 Akka 控制流量？

Question

我读到使用 Akka 时的一条重要规则是避免任何阻塞输入/输出操作、轮询、忙等待、休眠等。但是如果我真的需要一些流控制怎么办？

我正在使用 Akka Actor 向我们的客户发送邮件，并且为了对邮件服务器友好，每 5 秒发送一封邮件。我的计划是使用 dispatcher actor 来进行流控制，并使用 sender 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

在不阻塞的情况下执行此操作的一种方法是让另一个参与者随时保持发件箱状态。

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 演员成为scheduled actor，每 5 秒轮询一次发件箱。它向Outbox 参与者发送一条询问MailToSend 消息，如果响应为Some，则发送一封邮件。

看起来像这样：

调度发送方演员：

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

发送方演员：

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