0

我有将文件上传到 Dropbox 的父级 -> 子级角色关系。该关系由主管参与者和上传参与者组成。主管角色为上传角色定义主管策略。因此，如果上传到 Dropbox 失败，只要达到最大重试次数，就应该重新启动 actor。在我的应用程序中，我对上传的状态感兴趣。所以我使用询问模式来接收成功或失败的未来。您可以在下面找到我的演员的当前实现。

/**
 * An upload message.
 *
 * @param byte The byte array representing the content of a file.
 * @param path The path under which the file should be stored.
 */
case class UploadMsg(byte: Array[Byte], path: String)

/**
 * The upload supervisor.
 */
class UploadSupervisor extends Actor {

  /**
   * Stores the sender to the executing actor.
   */
  val senders: ParHashMap[String, ActorRef] = ParHashMap()

  /**
   * Defines the supervisor strategy
   */
  override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
    case _: DbxException => Restart
    case e: Exception => Stop
  }

  /**
   * Handles the received messages.
   */
  override def receive: Actor.Receive = {
    case msg: UploadMsg =>
      implicit val timeout = Timeout(60.seconds)

      val actor = context.actorOf(PropsContext.get(classOf[UploadActor]))
      senders += actor.path.toString -> sender
      context.watch(actor)
      ask(actor, msg).mapTo[Unit] pipeTo sender

    case Terminated(a) =>
      context.unwatch(a)
      senders.get(a.path.toString).map { sender =>
        sender ! akka.actor.Status.Failure(new Exception("Actor terminated"))
        senders - a.path.toString
      }
  }
}

/**
 * An aktor which uploads a file to Dropbox.
 */
class UploadActor @Inject() (client: DropboxClient) extends Actor {

  /**
   * Sends the message again after restart.
   *
   * @param reason The reason why an restart occurred.
   * @param message The message which causes the restart.
   */
  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    super.preRestart(reason, message)
    message foreach { self forward }
  }

  /**
   * Handles the received messages.
   */
  override def receive: Receive = {
    case UploadMsg(byte, path) =>
      val encrypted = encryptor.encrypt(byte)
      val is = new ByteArrayInputStream(encrypted)
      try {
        client.storeFile("/" + path, DbxWriteMode.force(), encrypted.length, is)
        sender ! (())
      } finally {
        is.close()
      }
  }
}

我现在的问题是：有没有更好的解决方案来告诉我的应用程序上传actor在指定次数或重试后失败。尤其是为演员存储发送者的地图，感觉有点别扭。

4

1 回答 1

1

你应该使用断路器

val breaker =
 new CircuitBreaker(context.system.scheduler,
  maxFailures = 5,
  callTimeout = 10.seconds,
  resetTimeout = 1.minute)

然后用断路器包装你的消息：

sender() ! breaker.withSyncCircuitBreaker(dangerousCall)

Breaker 具有三种状态：Closed、Open 和 HalfOpen。正常状态为关闭，当消息失败 $maxFailures 次状态更改为打开时。Breaker 为状态更改提供回调。如果您想做某事，请使用它。例如：

breaker onOpen { sender ! FailureMessage()}
于 2015-02-12T10:03:52.660 回答