Akka：保护异步资源

Question

虽然我的 SRP 方式的错误已得到纠正 yesterday，但我仍然想知道您如何干净地保证单线程访问 akka 中的异步资源，例如文件句柄。显然我不想允许从不同的线程对它分派多个读取和写入操作，但是如果我的参与者在该文件上调用基于未来的 API，这很可能会发生。

我想出的最好的模式是这样的：

trait AsyncIO {
  def Read(offset: Int, count: Int) : Future[ByteBuffer] = ???
}

object GuardedIOActor {
  case class Read(offset: Int, count: Int)
  case class ReadResult(data: ByteBuffer)
  private case class ReadCompleted()
}

class GuardedIOActor extends Actor with Stash with AsyncIO {
  import GuardedIOActor._
  var caller :Option[ActorRef] = None

  def receive = {
    case Read(offset,count) => caller match {
      case None => {
        caller = Some(sender)
        Read(offset,count).onSuccess({
          case data => {
            self ! ReadCompleted()
            caller.get ! ReadResult(data)
          }
        })
      }
      case Some(_) => {
        stash()
      }
    }
    case ReadCompleted() => {
      caller = None
      unstashAll()
    }
  }
}

但是这个要求对我来说还不够深奥。我的意思是应该有大量的资源需要同步访问但有一个异步 API。我是否忽略了一些常见的命名模式？

Answer 1

我认为您的解决方案的要点还不错，但是您可以使用 context.become 使您的演员表现得更像一个状态机：

class GaurdedIOActor extends Actor with Stash with AsyncIO {
  import GuardedIOActor._

  def receive = notReading

  def notReading: Receive = {
    case Read(offset, count) => {
      val caller = sender
      Read(offset,count).onSuccess({
        case data => {
          self ! ReadCompleted()
          caller ! ReadResult(data)
        }
      })
      context.become(reading)
    }
  }

  def reading: Receive = {
    case r: Read => stash()
    case ReadCompleted() => {
      context.become(notReading)
      unstashAll()
    }
  }
}

现在你的演员有两个明确定义的状态，不需要var

Answer 2

我意识到这个添加已经晚了一年，但是这个问题帮助我推理了我遇到的类似情况。以下 trait 封装了上面建议的功能，以减少参与者中的样板。它可以与任何带有 Stash 的演员混合。用法类似于pipeTo 模式；只需输入future.pipeSequentiallyTo(sender)，您的actor 将在future 完成并发送响应之前不会处理消息。

import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.language.implicitConversions
import scala.util.Failure
import scala.util.Success

import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.Stash
import akka.actor.Status

trait SequentialPipeToSupport { this: Actor with Stash =>

  case object ProcessingFinished

  def gotoProcessingState() = context.become(processingState)

  def markProcessingFinished() = self ! ProcessingFinished

  def processingState: Receive = {
    case ProcessingFinished =>
      unstashAll()
      context.unbecome()
    case _ =>
      stash()
  }

  final class SequentialPipeableFuture[T](val future: Future[T])(implicit executionContext: ExecutionContext) {

    def pipeSequentiallyTo(recipient: ActorRef): Future[T] = {

      gotoProcessingState()

      future onComplete {
        case Success(r) =>
          markProcessingFinished()
          recipient ! r
        case Failure(f) =>
          markProcessingFinished()
          recipient ! Status.Failure(f)
      }

      future
    }
  }

  implicit def pipeSequentially[T](future: Future[T])(implicit executionContext: ExecutionContext) = 
    new SequentialPipeableFuture(future)

}

也可作为Gist使用