容错示例

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

正常流程：

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

失败流程：

容错示例完整源代码" class="reference-link">容错示例完整源代码

1. import akka.actor._
2. import akka.actor.SupervisorStrategy._
3. import scala.concurrent.duration._
4. import akka.util.Timeout
5. import akka.event.LoggingReceive
6. import akka.pattern.{ ask, pipe }
7. import com.typesafe.config.ConfigFactory

9. /**
10. * Runs the sample
11. */
12. object FaultHandlingDocSample extends App {
13. import Worker._

15. val config = ConfigFactory.parseString("""
16. akka.loglevel = "DEBUG"
17. akka.actor.debug {
18. receive = on
19. lifecycle = on
20. }
21. """)

23. val system = ActorSystem("FaultToleranceSample", config)
24. val worker = system.actorOf(Props[Worker], name = "worker")
25. val listener = system.actorOf(Props[Listener], name = "listener")
26. // start the work and listen on progress
27. // note that the listener is used as sender of the tell,
28. // i.e. it will receive replies from the worker
29. worker.tell(Start, sender = listener)
30. }

32. /**
33. * Listens on progress from the worker and shuts down the system when enough
34. * work has been done.
35. */
36. class Listener extends Actor with ActorLogging {
37. import Worker._
38. // If we don't get any progress within 15 seconds then the service is unavailable
39. context.setReceiveTimeout(15 seconds)

41. def receive = {
42. case Progress(percent) =>
43. log.info("Current progress: {} %", percent)
44. if (percent >= 100.0) {
45. log.info("That's all, shutting down")
46. context.system.shutdown()
47. }

49. case ReceiveTimeout =>
50. // No progress within 15 seconds, ServiceUnavailable
51. log.error("Shutting down due to unavailable service")
52. context.system.shutdown()
53. }
54. }

56. object Worker {
57. case object Start
58. case object Do
59. case class Progress(percent: Double)
60. }

62. /**
63. * Worker performs some work when it receives the `Start` message.
64. * It will continuously notify the sender of the `Start` message
65. * of current ``Progress``. The `Worker` supervise the `CounterService`.
66. */
67. class Worker extends Actor with ActorLogging {
68. import Worker._
69. import CounterService._
70. implicit val askTimeout = Timeout(5 seconds)

72. // Stop the CounterService child if it throws ServiceUnavailable
73. override val supervisorStrategy = OneForOneStrategy() {
74. case _: CounterService.ServiceUnavailable => Stop
75. }

77. // The sender of the initial Start message will continuously be notified
78. // about progress
79. var progressListener: Option[ActorRef] = None
80. val counterService = context.actorOf(Props[CounterService], name = "counter")
81. val totalCount = 51
82. import context.dispatcher // Use this Actors' Dispatcher as ExecutionContext

84. def receive = LoggingReceive {
85. case Start if progressListener.isEmpty =>
86. progressListener = Some(sender)
87. context.system.scheduler.schedule(Duration.Zero, 1 second, self, Do)

89. case Do =>
90. counterService ! Increment(1)
91. counterService ! Increment(1)
92. counterService ! Increment(1)

94. // Send current progress to the initial sender
95. counterService ? GetCurrentCount map {
96. case CurrentCount(_, count) => Progress(100.0 * count / totalCount)
97. } pipeTo progressListener.get
98. }
99. }

101. object CounterService {
102. case class Increment(n: Int)
103. case object GetCurrentCount
104. case class CurrentCount(key: String, count: Long)
105. class ServiceUnavailable(msg: String) extends RuntimeException(msg)

107. private case object Reconnect
108. }

110. /**
111. * Adds the value received in `Increment` message to a persistent
112. * counter. Replies with `CurrentCount` when it is asked for `CurrentCount`.
113. * `CounterService` supervise `Storage` and `Counter`.
114. */
115. class CounterService extends Actor {
116. import CounterService._
117. import Counter._
118. import Storage._

120. // Restart the storage child when StorageException is thrown.
121. // After 3 restarts within 5 seconds it will be stopped.
122. override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 3,
123. withinTimeRange = 5 seconds) {
124. case _: Storage.StorageException => Restart
125. }

127. val key = self.path.name
128. var storage: Option[ActorRef] = None
129. var counter: Option[ActorRef] = None
130. var backlog = IndexedSeq.empty[(ActorRef, Any)]
131. val MaxBacklog = 10000

133. import context.dispatcher // Use this Actors' Dispatcher as ExecutionContext

135. override def preStart() {
136. initStorage()
137. }

139. /**
140. * The child storage is restarted in case of failure, but after 3 restarts,
141. * and still failing it will be stopped. Better to back-off than continuously
142. * failing. When it has been stopped we will schedule a Reconnect after a delay.
143. * Watch the child so we receive Terminated message when it has been terminated.
144. */
145. def initStorage() {
146. storage = Some(context.watch(context.actorOf(Props[Storage], name = "storage")))
147. // Tell the counter, if any, to use the new storage
148. counter foreach { _ ! UseStorage(storage) }
149. // We need the initial value to be able to operate
150. storage.get ! Get(key)
151. }

153. def receive = LoggingReceive {

155. case Entry(k, v) if k == key && counter == None =>
156. // Reply from Storage of the initial value, now we can create the Counter
157. val c = context.actorOf(Props(classOf[Counter], key, v))
158. counter = Some(c)
159. // Tell the counter to use current storage
160. c ! UseStorage(storage)
161. // and send the buffered backlog to the counter
162. for ((replyTo, msg) <- backlog) c.tell(msg, sender = replyTo)
163. backlog = IndexedSeq.empty

165. case msg @ Increment(n)    => forwardOrPlaceInBacklog(msg)

167. case msg @ GetCurrentCount => forwardOrPlaceInBacklog(msg)

169. case Terminated(actorRef) if Some(actorRef) == storage =>
170. // After 3 restarts the storage child is stopped.
171. // We receive Terminated because we watch the child, see initStorage.
172. storage = None
173. // Tell the counter that there is no storage for the moment
174. counter foreach { _ ! UseStorage(None) }
175. // Try to re-establish storage after while
176. context.system.scheduler.scheduleOnce(10 seconds, self, Reconnect)

178. case Reconnect =>
179. // Re-establish storage after the scheduled delay
180. initStorage()
181. }

183. def forwardOrPlaceInBacklog(msg: Any) {
184. // We need the initial value from storage before we can start delegate to
185. // the counter. Before that we place the messages in a backlog, to be sent
186. // to the counter when it is initialized.
187. counter match {
188. case Some(c) => c forward msg
189. case None =>
190. if (backlog.size >= MaxBacklog)
191. throw new ServiceUnavailable(
192. "CounterService not available, lack of initial value")
193. backlog :+= (sender() -> msg)
194. }
195. }

197. }

199. object Counter {
200. case class UseStorage(storage: Option[ActorRef])
201. }

203. /**
204. * The in memory count variable that will send current
205. * value to the `Storage`, if there is any storage
206. * available at the moment.
207. */
208. class Counter(key: String, initialValue: Long) extends Actor {
209. import Counter._
210. import CounterService._
211. import Storage._

213. var count = initialValue
214. var storage: Option[ActorRef] = None

216. def receive = LoggingReceive {
217. case UseStorage(s) =>
218. storage = s
219. storeCount()

221. case Increment(n) =>
222. count += n
223. storeCount()

225. case GetCurrentCount =>
226. sender() ! CurrentCount(key, count)

228. }

230. def storeCount() {
231. // Delegate dangerous work, to protect our valuable state.
232. // We can continue without storage.
233. storage foreach { _ ! Store(Entry(key, count)) }
234. }

236. }

238. object Storage {
239. case class Store(entry: Entry)
240. case class Get(key: String)
241. case class Entry(key: String, value: Long)
242. class StorageException(msg: String) extends RuntimeException(msg)
243. }

245. /**
246. * Saves key/value pairs to persistent storage when receiving `Store` message.
247. * Replies with current value when receiving `Get` message.
248. * Will throw StorageException if the underlying data store is out of order.
249. */
250. class Storage extends Actor {
251. import Storage._

253. val db = DummyDB

255. def receive = LoggingReceive {
256. case Store(Entry(key, count)) => db.save(key, count)
257. case Get(key)                 => sender() ! Entry(key, db.load(key).getOrElse(0L))
258. }
259. }

261. object DummyDB {
262. import Storage.StorageException
263. private var db = Map[String, Long]()

265. @throws(classOf[StorageException])
266. def save(key: String, value: Long): Unit = synchronized {
267. if (11 <= value && value <= 14)
268. throw new StorageException("Simulated store failure " + value)
269. db += (key -> value)
270. }

272. @throws(classOf[StorageException])
273. def load(key: String): Option[Long] = synchronized {
274. db.get(key)
275. }
276. }