zoukankan      html  css  js  c++  java
  • 顺序执行到来的消息 actor

    在某项目里,有个 actor 需要做一些持久化的操作,这些操作耗时比较久,理应使用异步的代码来写,但是需求又强调每次只能做一个持久化操作,后来的请求应该等待。一个显然的做法是阻塞式的写,这样就能比较简单的实现顺序花操作。

    代码写完以后,我觉得在 actor 中 block 不够完美,就想其他的解决方案。实际上,借助 akka actor 的一些函数,可以实现在不阻塞的情况下实现顺序执行请求的功能的。这种办法的核心是使用 become, unbecome 函数:actor 设置两种状态 free 和 busy,当 free 的时,处理消息,当 busy 时,暂时将消息存储起来,处理消息后,给 actor 返回 done 指令,actor 的状态重新返回到 free,准备处理下一个请求。具体的实现又有很多细节可以考虑,比如当 busy 时到来的请求存储到哪里,是 stash 起来还是在 actor 内部维护一个 queue。请求的处理逻辑是写在 actor 内部,还是借鉴 cameo 模式,再创建一个 actor。

    网上已有一种实现,我看了下,觉得应该没有问题。只不过 actor 内部维护了一个 queue,这可能会造成 actor 死亡后重启数据丢失的情况。更好的办法应该是 cameo 模式创建新的 actor 来处理可能出现异常(危险)的工作,其次是把 actor 的 mailbox 当做那个 queue,不要自己维护,按照 doc 缩写,actor 重启后,mailbox 的消息不会丢失。

    package actors
    
    import scala.concurrent.Future
    import scala.concurrent.ExecutionContext.Implicits.global
    
    import akka.actor.{ Actor, ActorRef }
    
    import play.api.libs.concurrent.Akka
    import play.api.Logger
    import play.api.Play.current
    
    trait SequentialActor {
      this: Actor =>
    
      import SequentialActor._
    
      // Actor defines type Receive as PartialFunction[Any, Unit]
      type ReceiveAsync = PartialFunction[Any, Future[_]]
    
      private val queue = scala.collection.mutable.Queue[Job]()
      private def enqueue(job: Job): Unit = queue enqueue job
      private def dequeue: Option[Job] = if (queue.isEmpty) None else Some(queue.dequeue)
    
      private var _senderAsync: ActorRef = _
      def senderAsync = _senderAsync
    
      def receive: Receive = {
        case msg =>
          context become busy
          process(Job(msg, sender))
      }
    
      def busy: Receive = {
        case Done =>
          dequeue match {
            case None => context.unbecome
            case Some(job) => process(job)
          }
        case msg =>
          enqueue(Job(msg, sender))
      }
    
      def process(job: Job) {
        _senderAsync = job.sender
        (receiveAsync orElse fallback)(job.msg).onComplete { _ => self ! Done }
      }
    
      def receiveAsync: ReceiveAsync
    
      def fallback: ReceiveAsync = {
        case msg =>
          Logger.error(s"Unhandled message: $msg")
          Future.successful{ () }
      }
    
    }
    object SequentialActor {
      case object Done
      case class Job(msg: Any, sender: ActorRef)
    }
    

      

  • 相关阅读:
    jQuery easyUI 的combogrid进行模糊匹配
    SQL 快速生成不重复的卡号
    用SQL语句查找包含有某个关键字的存储过程、触发器、函数等等
    C# DateTime日期格式化
    【SQL触发器】类型 FOR 、AFTER、 Instead of
    【BAT】批量给制定类型的文件名添加前缀
    【VBA】将Excel数据转化为txt文本数据
    【Python】将多个工作簿中的数据按列合并到同一个工作表
    【Python】从DB2数据库中取出字段信息并根据字段类型生成SQL
    【Oracle】多行数据合并为一行,使用逗号分隔
  • 原文地址:https://www.cnblogs.com/xinsheng/p/4723277.html
Copyright © 2011-2022 走看看