Множественные вызовы Future в методе получения Актера

Не зная слишком много о клиенте Redis, который вы используете, я могу предложить альтернативное решение, которое должно быть чище и не будет иметь проблем с закрытием изменяемого состояния. Идея состоит в том, чтобы использовать ситуацию типа мастер/рабочий, когда мастер (SimpleRedisActor) получает запрос на выполнение работы, а затем делегирует его рабочему процессу, который выполняет работу и отвечает состоянием для обновления. Это решение будет выглядеть примерно так:

object SimpleRedisActor{
  case object UpdateState
  def props(ip:String, key:String) = Props(classOf[SimpleRedisActor], ip, key)
}

class SimpleRedisActor(ip: String, key: String) extends Actor with ActorLogging {
  import SimpleRedisActor._
  import SimpleRedisWorker._

  //mutable state that is updated on a periodic basis
  var mutableState: Set[String] = Set.empty

  val rClient = RedisClient(ip)(context.system)

  def receive = {
    case UpdateState => 
      log.info("Start of UpdateState ...")      
      val worker = context.actorOf(SimpleRedisWorker.props)
      worker ! DoWork(rClient, key)

    case WorkResult(result) =>
      mutableState ++ result

    case FailedWorkResult(ex) =>
      log.error("Worker got failed work result", ex)
  }
}

object SimpleRedisWorker{
  case class DoWork(client:RedisClient, key:String)
  case class WorkResult(result:Seq[String])
  case class FailedWorkResult(ex:Throwable)
  def props = Props[SimpleRedisWorker]
}

class SimpleRedisWorker extends Actor{
  import SimpleRedisWorker._
  import akka.pattern.pipe
  import context._

  def receive = {
    case DoWork(client, key) =>
      val trans = client.transaction()
      trans.zcard(key) pipeTo self
      become(waitingForZCard(sender, trans, key) orElse failureHandler(sender, trans))
  }

  def waitingForZCard(orig:ActorRef, trans:RedisTransaction, key:String):Receive = {      
    case l:Long =>
      trans.zrange(key, l -1, l) pipeTo self
      become(waitingForZRange(orig, trans) orElse failureHandler(orig, trans))
  }

  def waitingForZRange(orig:ActorRef, trans:RedisTransaction):Receive = {
    case s:Seq[ByteString] =>
      orig ! WorkResult(s.map(_.utf8String))
      finishAndStop(trans)
  }

  def failureHandler(orig:ActorRef, trans:RedisTransaction):Receive = {
    case Status.Failure(ex) => 
      orig ! FailedWorkResult(ex)
      finishAndStop(trans)   
  }

  def finishAndStop(trans:RedisTransaction) {
    trans.exec()
    context stop self
  }
}

Рабочий начинает транзакцию, затем вызывает Redis и в конечном итоге завершает транзакцию, прежде чем остановиться. Когда он вызывает redis, он получает будущее и передает его обратно самому себе для продолжения обработки, меняя метод получения между ними как механизм отображения прогресса в его состояниях. В такой модели (которая, как я полагаю, чем-то похожа на шаблон ядра ошибки) мастер владеет состоянием и защищает его, делегируя «рискованную» работу ребенку, который может понять, каким должно быть изменение состояния. но изменение по-прежнему принадлежит хозяину.

Опять же, я понятия не имею о возможностях клиента Redis, который вы используете, и достаточно ли он безопасен, чтобы делать такие вещи, но это не совсем так. Цель состояла в том, чтобы показать более безопасную структуру для выполнения чего-то подобного, включающего фьючерсы и состояния, которые необходимо безопасно изменить.

Использование обратного вызова для изменения внутреннего состояния не является хорошей идеей, выдержка из документы akka:

При использовании будущих обратных вызовов, таких как onComplete, onSuccess и onFailure, внутри акторов необходимо тщательно избегать закрытия ссылки на содержащий актор, т. е. не вызывать методы или обращаться к изменяемому состоянию включающего актора из обратного вызова.

Почему вы беспокоитесь о pipeTo и транзакциях? Не уверен, как работают транзакции Redis, но я предполагаю, что транзакция в любом случае не включает обратный вызов onComplete во втором будущем.

Я бы вынес государство в отдельный актор, которому вы тоже трубите будущее. Таким образом, у вас есть отдельный почтовый ящик, и порядок в нем будет таким же, как и порядок сообщений, пришедших для изменения состояния. Кроме того, если поступают какие-либо запросы на чтение, они также будут помещены в правильном порядке.

Редактировать, чтобы ответить на отредактированный вопрос: Итак, вы не хотите передавать первое будущее, это имеет смысл и не должно быть проблемой, поскольку первый обратный вызов безвреден. Обратный вызов второго будущего представляет собой проблему, поскольку он манипулирует состоянием. Но это будущее может быть реализовано без необходимости доступа к транзакции.

Итак, в основном мое предложение:

val firstFuture = tran.zcard
firstFuture.onComplete {
   val secondFuture = tran.zrange
   secondFuture pipeTo stateActor
}

С stateActor, содержащим изменяемое состояние.