在特定时间可靠地处理消息

Question

我们假设我有一个聊天应用程序.

客户端向聊天发送消息,导致某些Actor的某些命令.现在,我想立即处理他写的内容,并在此聊天中将其提供给其他用户,因此我处理此命令.同时我想告诉自己(一个演员)我需要在聊天记录数据库中存储此消息但不是现在.保存到数据库应该每2分钟发生一次.如果发生崩溃,我应该能够坚持到数据库.

我假设工作流程是这样的:

1. 用户发送消息
2. 聊天室演员收到了带有此消息的命令
3. 我们向所有人广播此消息,并将此消息添加到某种队列以将其保留到聊天历史数据库
4. 一些持久命令在超过2分钟超时时运行.它会按照到达顺序收集尚未保留的所有传入聊天消息
5. 使用所有消息运行事务,然后将其从队列中删除.
6. 如果在3之后的某个地方发生了崩溃并且消息没有持续存在,那么我应该尝试再次坚持它们.如果坚持下去,我就不应该再试图坚持下去.

如何在Akka中构建这样的东西？我应该使用哪些功能/哪种模式？

Answer 1

您可能需要两个演员:一个(协调员)将向客户端发送有关聊天命令的通知.另一个(节流器) - 每2分钟将数据推送到数据库.您的队列将只是一个内部状态的节流器:

class Coordinator extends Actor {
   def receive = {
     case command: Record => 
          broadcast(command)
          throttler ! command
   }
}


class Throttler extends Actor {

  import system.dispatcher

  val queue = mutable.List[Record] //it may be a cache instead

  def schedule = system.scheduler.scheduleOnce(2 minutes, self, Tick) // http://doc.akka.io/docs/akka/snapshot/scala/scheduler.html


  def receive = {
       case Start => schedule
       case command: Record =>
           queue ++= command
       case Tick => 
          schedule
          try {
            //---open transaction here---
            for (r <- queue) push(r)
            //---close transaction here---
            queue.clear //will not be cleared in case of exception
          } catch {...}
  }
}

您也可以使用基于FSM的实现,如@abatyuk所说.

如果您需要减少邮箱的负载 - 您可以尝试一些背压/负载平衡模式,如Akka Work Pulling.

如果要保护节点本身(如果某些服务器节点发生故障,则恢复队列状态) - 您可以使用Akka Cluster复制(手动)队列的状态.在这种情况下,协调员应该是Cluster Singleton并且应该自己向随机actor发送ticks(你可以使用总线),并作为主管保持他们的成功和失败.请注意,管理员状态可能会丢失,因此您还应该通过节点复制它(并且每隔2分钟在它们之间合并状态,因此最好SortedSet用于队列,因为合并将类似sets.reduce(_ ++ _)).

像Riak这样的存储已经提供了解决集群化问题的简单方法,因此您可以将它们用作队列(协调器和节流器都将是"无状态"单例).对于Riak,您可以将其配置为Available + Partitioning(请参阅CAP定理),因为合并数据不是问题 - 您的聊天记录是CRDT(无冲突)数据类型.

另一种解决方案是具有WriteBehind模式的缓存(配置为每2分钟启动一次)作为调节器.

事件采购也可以保护你的演员的状态,但是当你需要在恢复后重做所有动作时它更有用(你不需要它 - 它会重新提交一切到数据库).您可以使用快照(它与直接使用缓存非常相似),但如果您关心可用性,最好将它们保存到缓存(通过实施SnapshotStore)而不是本地FS.请注意,您还可能必须删除以前保存的快照以减小存储大小.并且您应该同步保存每条记录以避免失去状态.

PS不要忘记向发件人(对您的javascript)确认邮件,或者即使将缓存作为队列,您也可以丢失最后的邮件(在邮箱中).

PS/2数据库对于actor的状态持久性几乎总是一个糟糕的解决方案,因为它很慢并且可能变得不可用.我不会推荐像MongoDB这样强大一致的NoSQL解决方案 - 最终的一致性是您的最佳选择.