scala – 阿卡事件总线教程

不知道是否有或没有任何好的教程,但我可以给你一个可能的用户案例的快速例子,使用事件流可能是有帮助的.但是,在高级别上,事件流是满足您的应用可能具有的pub / sub类型要求的良好机制.假设您有一个用例来更新系统中的用户余额.平衡经常被访问,所以你决定缓存它以获得更好的性能.当余额更新时,您还需要查看用户是否跨越了余额的门槛,如果是这样,请发送电子邮件.您不想将缓存丢弃或平衡阈值检查直接绑定到主平衡更新呼叫中,因为它们可能重量很大,并且减慢用户的响应.您可以对这样的特定要求进行建模：

//Message and event classes
case class UpdateAccountBalance(userId:Long, amount:Long)
case class BalanceUpdated(userId:Long)

//Actor that performs account updates
class AccountManager extends Actor{
  val dao = new AccountManagerDao

  def receive = {
    case UpdateAccountBalance(userId, amount) =>
      val res = for(result <- dao.updateBalance(userId, amount)) yield{
        context.system.eventStream.publish(BalanceUpdated(userId))
        result                
      }

      sender ! res
  }
}

//Actor that manages a cache of account balance data
class AccountCacher extends Actor{
  val cache = new AccountCache

  override def preStart = {
    context.system.eventStream.subscribe(context.self, classOf[BalanceUpdated])
  }

  def receive = {
    case BalanceUpdated(userId) =>
      cache.remove(userId)
  }
}

//Actor that checks balance after an update to warn of low balance
class LowBalanceChecker extends Actor{
  val dao = new LowBalanceDao

  override def preStart = {
    context.system.eventStream.subscribe(context.self, classOf[BalanceUpdated])
  }

  def receive = {
    case BalanceUpdated(userId) =>
      for{
        balance <- dao.getBalance(userId)
        theshold <- dao.getBalanceThreshold(userId)
        if (balance < threshold)
      }{
        sendBalanceEmail(userId, balance)
      }
  }
}

在这个例子中,AccountCacher和LowBalanceChecker参与者根据BalanceUpdated事件的类类型来订阅eventStream.如果事件发布到流,则将由两个这些actor实例接收.然后,在AccountManager中,当余额更新成功时,会为用户引发一个BalanceUpdated事件.当这种情况发生时,并行地将该消息传递给AccountCacher和LowBalanceChecker的邮箱,导致余额从缓存中删除并检查帐户阈值,并且可能发送电子邮件.

现在,您可以直接将(！)调用到AccountManager中直接与其他两个演员进行通信,但是可以认为可能会太平滑地将这两种“副作用”耦合到平衡更新中,而这些类型的细节不一定属于AccountManager.如果您有一些条件可能会导致一些额外的事情(检查,更新等),需要纯粹作为副作用发生(不是核心业务流程本身的一部分),则事件流可能是一种很好的方式解除事件的发生和谁可能需要对事件做出反应.