• Introduction
• 1. 引言
  • 1.1. Akka是什么?
  • 1.2. 为什么使用Akka?
  • 1.3. 入门
  • 1.4. 必修的“Hello World”
  • 1.5. 用例和部署场景
  • 1.6. Akka使用实例
• 2. 概述
  • 2.1. 术语，概念
  • 2.2. Actor系统
  • 2.3. 什么是Actor?
  • 2.4. 监管与监控
  • 2.5. Actor引用, 路径与地址
  • 2.6. 位置透明性
  • 2.7. Akka与Java内存模型
  • 2.8. 消息发送语义
  • 2.9. 配置
• 3. Actors
  • 3.1. Actors
  • 3.2. 有类型Actor
  • 3.3. 容错
    • 3.3.1. 容错示例
  • 3.4. 调度器
  • 3.5. 邮箱
  • 3.6. 路由
  • 3.7. 有限状态机(FSM)
  • 3.8. 持久化
  • 3.9. 测试Actor系统
    • 3.9.1. TestKit实例
  • 3.10. Actor DSL
• 4. Futures与Agents
  • 4.1. Futures
  • 4.2. Agents
• 5. 网络
  • 5.1. 集群规格
  • 5.2. 集群用法
  • 5.3. 远程
  • 5.4. 序列化
  • 5.5. I/O
  • 5.6. 使用TCP
  • 5.7. 使用UDP
  • 5.8. ZeroMQ
  • 5.9. Camel
• 6. 实用工具
  • 6.1. 事件总线
  • 6.2. 日志
  • 6.3. 调度器
  • 6.4. Duration
  • 6.5. 线路断路器
  • 6.6. Akka扩展
  • 6.7. 微内核
• 7. 如何使用：常用模式
• 8. 实验模块
  • 8.1. 持久化
  • 8.2. 多节点测试
  • 8.3. Actors(使用Java的Lambda支持)
  • 8.4. FSM(使用Java的Lambda支持)
  • 8.5. 外部贡献
• 9. Akka开发者信息
  • 9.1. 构建Akka
  • 9.2. 多JVM测试
  • 9.3. I/O层设计
  • 9.4. 开发指南
  • 9.5. 文档指南
  • 9.6. 团队
• 10. 工程信息
  • 10.1. 迁移指南
  • 10.2. 问题追踪
  • 10.3. 许可证
  • 10.4. 赞助商
  • 10.5. 项目
• 11. 附加信息
  • 11.1. 常见问题
  • 11.2. 图书
  • 11.3. 其他语言绑定
  • 11.4. Akka与OSGi
  • 11.5. 部分HTTP框架名单
• Published using GitBook

AKKA 2.3.6 Scala 文档

容错示例

以上图示了正常的消息流。

正常流程：

以上图示了当发生数据库失败时的过程

失败流程：

容错示例完整源代码

import akka.actor._
import akka.actor.SupervisorStrategy._
import scala.concurrent.duration._
import akka.util.Timeout
import akka.event.LoggingReceive
import akka.pattern.{ ask, pipe }
import com.typesafe.config.ConfigFactory
/**
 * Runs the sample
 */
object FaultHandlingDocSample extends App {
  import Worker._
  val config = ConfigFactory.parseString("""
    akka.loglevel = "DEBUG"
    akka.actor.debug {
      receive = on
      lifecycle = on
    }
    """)
  val system = ActorSystem("FaultToleranceSample", config)
  val worker = system.actorOf(Props[Worker], name = "worker")
  val listener = system.actorOf(Props[Listener], name = "listener")
  // start the work and listen on progress
  // note that the listener is used as sender of the tell,
  // i.e. it will receive replies from the worker
  worker.tell(Start, sender = listener)
}
/**
 * Listens on progress from the worker and shuts down the system when enough
 * work has been done.
 */
class Listener extends Actor with ActorLogging {
  import Worker._
  // If we don't get any progress within 15 seconds then the service is unavailable
  context.setReceiveTimeout(15 seconds)
  def receive = {
    case Progress(percent) =>
      log.info("Current progress: {} %", percent)
      if (percent >= 100.0) {
        log.info("That's all, shutting down")
        context.system.shutdown()
      }
    case ReceiveTimeout =>
      // No progress within 15 seconds, ServiceUnavailable
      log.error("Shutting down due to unavailable service")
      context.system.shutdown()
  }
}
object Worker {
  case object Start
  case object Do
  case class Progress(percent: Double)
}
/**
 * Worker performs some work when it receives the `Start` message.
 * It will continuously notify the sender of the `Start` message
 * of current ``Progress``. The `Worker` supervise the `CounterService`.
 */
class Worker extends Actor with ActorLogging {
  import Worker._
  import CounterService._
  implicit val askTimeout = Timeout(5 seconds)
  // Stop the CounterService child if it throws ServiceUnavailable
  override val supervisorStrategy = OneForOneStrategy() {
    case _: CounterService.ServiceUnavailable => Stop
  }
  // The sender of the initial Start message will continuously be notified
  // about progress
  var progressListener: Option[ActorRef] = None
  val counterService = context.actorOf(Props[CounterService], name = "counter")
  val totalCount = 51
  import context.dispatcher // Use this Actors' Dispatcher as ExecutionContext
  def receive = LoggingReceive {
    case Start if progressListener.isEmpty =>
      progressListener = Some(sender)
      context.system.scheduler.schedule(Duration.Zero, 1 second, self, Do)
    case Do =>
      counterService ! Increment(1)
      counterService ! Increment(1)
      counterService ! Increment(1)
      // Send current progress to the initial sender
      counterService ? GetCurrentCount map {
        case CurrentCount(_, count) => Progress(100.0 * count / totalCount)
      } pipeTo progressListener.get
  }
}
object CounterService {
  case class Increment(n: Int)
  case object GetCurrentCount
  case class CurrentCount(key: String, count: Long)
  class ServiceUnavailable(msg: String) extends RuntimeException(msg)
  private case object Reconnect
}
/**
 * Adds the value received in `Increment` message to a persistent
 * counter. Replies with `CurrentCount` when it is asked for `CurrentCount`.
 * `CounterService` supervise `Storage` and `Counter`.
 */
class CounterService extends Actor {
  import CounterService._
  import Counter._
  import Storage._
  // Restart the storage child when StorageException is thrown.
  // After 3 restarts within 5 seconds it will be stopped.
  override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 3,
    withinTimeRange = 5 seconds) {
      case _: Storage.StorageException => Restart
    }
  val key = self.path.name
  var storage: Option[ActorRef] = None
  var counter: Option[ActorRef] = None
  var backlog = IndexedSeq.empty[(ActorRef, Any)]
  val MaxBacklog = 10000
  import context.dispatcher // Use this Actors' Dispatcher as ExecutionContext
  override def preStart() {
    initStorage()
  }
  /**
   * The child storage is restarted in case of failure, but after 3 restarts,
   * and still failing it will be stopped. Better to back-off than continuously
   * failing. When it has been stopped we will schedule a Reconnect after a delay.
   * Watch the child so we receive Terminated message when it has been terminated.
   */
  def initStorage() {
    storage = Some(context.watch(context.actorOf(Props[Storage], name = "storage")))
    // Tell the counter, if any, to use the new storage
    counter foreach { _ ! UseStorage(storage) }
    // We need the initial value to be able to operate
    storage.get ! Get(key)
  }
  def receive = LoggingReceive {
    case Entry(k, v) if k == key && counter == None =>
      // Reply from Storage of the initial value, now we can create the Counter
      val c = context.actorOf(Props(classOf[Counter], key, v))
      counter = Some(c)
      // Tell the counter to use current storage
      c ! UseStorage(storage)
      // and send the buffered backlog to the counter
      for ((replyTo, msg) <- backlog) c.tell(msg, sender = replyTo)
      backlog = IndexedSeq.empty
    case msg @ Increment(n)    => forwardOrPlaceInBacklog(msg)
    case msg @ GetCurrentCount => forwardOrPlaceInBacklog(msg)
    case Terminated(actorRef) if Some(actorRef) == storage =>
      // After 3 restarts the storage child is stopped.
      // We receive Terminated because we watch the child, see initStorage.
      storage = None
      // Tell the counter that there is no storage for the moment
      counter foreach { _ ! UseStorage(None) }
      // Try to re-establish storage after while
      context.system.scheduler.scheduleOnce(10 seconds, self, Reconnect)
    case Reconnect =>
      // Re-establish storage after the scheduled delay
      initStorage()
  }
  def forwardOrPlaceInBacklog(msg: Any) {
    // We need the initial value from storage before we can start delegate to
    // the counter. Before that we place the messages in a backlog, to be sent
    // to the counter when it is initialized.
    counter match {
      case Some(c) => c forward msg
      case None =>
        if (backlog.size >= MaxBacklog)
          throw new ServiceUnavailable(
            "CounterService not available, lack of initial value")
        backlog :+= (sender() -> msg)
    }
  }
}
object Counter {
  case class UseStorage(storage: Option[ActorRef])
}
/**
 * The in memory count variable that will send current
 * value to the `Storage`, if there is any storage
 * available at the moment.
 */
class Counter(key: String, initialValue: Long) extends Actor {
  import Counter._
  import CounterService._
  import Storage._
  var count = initialValue
  var storage: Option[ActorRef] = None
  def receive = LoggingReceive {
    case UseStorage(s) =>
      storage = s
      storeCount()
    case Increment(n) =>
      count += n
      storeCount()
    case GetCurrentCount =>
      sender() ! CurrentCount(key, count)
  }
  def storeCount() {
    // Delegate dangerous work, to protect our valuable state.
    // We can continue without storage.
    storage foreach { _ ! Store(Entry(key, count)) }
  }
}
object Storage {
  case class Store(entry: Entry)
  case class Get(key: String)
  case class Entry(key: String, value: Long)
  class StorageException(msg: String) extends RuntimeException(msg)
}
/**
 * Saves key/value pairs to persistent storage when receiving `Store` message.
 * Replies with current value when receiving `Get` message.
 * Will throw StorageException if the underlying data store is out of order.
 */
class Storage extends Actor {
  import Storage._
  val db = DummyDB
  def receive = LoggingReceive {
    case Store(Entry(key, count)) => db.save(key, count)
    case Get(key)                 => sender() ! Entry(key, db.load(key).getOrElse(0L))
  }
}
object DummyDB {
  import Storage.StorageException
  private var db = Map[String, Long]()
  @throws(classOf[StorageException])
  def save(key: String, value: Long): Unit = synchronized {
    if (11 <= value && value <= 14)
      throw new StorageException("Simulated store failure " + value)
    db += (key -> value)
  }
  @throws(classOf[StorageException])
  def load(key: String): Option[Long] = synchronized {
    db.get(key)
  }
}