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

      

  • 相关阅读:
    JB开发之二 [jailbreak,越狱开发研究]
    iOS9 Https技术预研
    Tweak和app交互方案【进程通信】
    iOS设备抓包终极解决方案(支持https)
    Anti-Anti dylib(反 反-dylib钩子(Anti-tweak))
    Hook ptrace 调试加入了ptrace函数的程序
    How do I determine if I'm being run under the debugger?
    个人整理的一些iOS Entitlements
    SQLite在多线程环境下的应用
    别的程序员是怎么读你的简历的
  • 原文地址:https://www.cnblogs.com/xinsheng/p/4723277.html
Copyright © 2011-2022 走看看