[Akka]CircuitBreakerはどう動くのか


# CircuitBreakerとは

原典にあたるのは[Martin Fowler氏のpost](http://martinfowler.com/bliki/CircuitBreaker.html)になるはず。
リモートへのリクエストが失敗した時にうまいこと障害が伝搬しないようにするためのデザイン。

昨今のMicroserviceアーキテクチャな文脈だと考えやすい。
特定のサービスがダウンしてしまった時に、
それに依存するサービスが引きずられてダウンしてしまわないようにするもの。
Reactiveなシステムを作る上で欠かせない存在になりそう。

## CircuitBreakerの状態

監視対象とするサービスの状態に応じてClose, Open, Half-Openの3つをとる。
それぞれの状態の意味はおおざっぱに以下。

- Close
  - 正常に動作していて、対象サービスにリクエストを送る状態
- Open
  - 対象サービスに異常があり、リクエストを送らない状態
- Half-Open
  - そろそろ直ったかな、とちょっとリクエストを送ろうとしてみる状態

Closeが正常でOpenが異常というのは何となく直感に反するが、
`CircuitBreaker`の語源が電気回路にあることを考慮すれば自然。

Close状態でサービスの失敗が重なるとOpenに移行、一定時間後にHalf-Openとなり、
サービスに一度リクエストを送信して成功したらClose、失敗したらOpenに戻る、という状態遷移。

# AkkaのCircuitBreaker

AkkaもCircuitBreakerを提供している。
[Circuit Breaker — Akka Documentation](http://doc.akka.io/docs/akka/current/common/circuitbreaker.html)
ちなみに[実装はここ](https://github.com/akka/akka/blob/master/akka-actor/src/main/scala/akka/pattern/CircuitBreaker.scala)にある。
この記事では`akka.pattern.CircuitBreaker`を使ってCircuitBreakerの動作を追って理解する。

# 動作を確認するためにサンプルを実装してみる

コードは[github](https://github.com/petitviolet/circuit-breaker-prac/blob/master/src/main/scala/com/example/AkkaCircuitBreakerPrac.scala)に置いた。

## サービスにあたるものを実装

まず、障害が起こりうるサービスをActorで用意する。

```scala
case class Message(value: String)
case object PanicMessage
case object HeavyMessage

class UnstableActor extends Actor {
  override def receive: Receive = {
    case Message(value) =>
      println(value)
      sender ! s"receive: $value"
    case PanicMessage =>
      // just fail
      sys.error("Oops...")
    case HeavyMessage =>
      // sleep over callTimeout
      Thread.sleep(2000)
      sender ! s"finish $HeavyMessage"
  }
}
```

- `Message(value)`は正常処理出来るメッセージ
- `PanicMessage`は例外をthrowしてしまうメッセージ
- `HeavyMessage`は時間のかかる処理をするメッセージ

となっている。

## CircuitBreakerを用意

```scala
implicit val system = ActorSystem(s"AkkaCircuitBreakerPrac")
val circuitBreaker = CircuitBreaker(
  system.scheduler,
  maxFailures = 2,
  callTimeout =  1.seconds,
  resetTimeout =  3.seconds)
  .onOpen(println(s"OPEN"))
  .onClose(println(s"CLOSE"))
  .onHalfOpen(println(s"HALF-OPEN"))
```

状態遷移時のコールバックを登録できるので標準出力に表示するようにしている。
`CircuitBreaker.apply`の引数それぞれの意味は以下。

- maxFailures
  - Close状態で`maxFailures`回、サービスの実行に失敗したらOpen状態に移行する
- callTimeout
  - サービスを実行しようとして`callTimeout`時間経過しても結果が返ってこなければ失敗とみなす
- resetTimeout
  - Open状態で`resetTimeout`時間経過したらHalf-Openに移行する

## CircuitBreakerの動作を見る

まずは`ActorRef`とかの準備

```scala
val actorRef = system.actorOf(Props[UnstableActor])
implicit val dispatcher: ExecutionContext = ExecutionContext.Implicits.global
implicit val timeout = Timeout(5.seconds)
```

`CircuitBreaker#withCircuitBreaker`で監視しつつ`ActorRef`への`?`を実行。
初期状態ではCloseになっているので正常に処理される。

```scala
// 状態はClose
circuitBreaker.withCircuitBreaker(actorRef ? Message("1"))
circuitBreaker.withCircuitBreaker(actorRef ? Message("2"))
```

次に一度`Failure`になる処理を実行する。
`maxFailure`が2なのでCloseなまま。

```scala
// 1回失敗
circuitBreaker.withCircuitBreaker(actorRef ? PanicMessage)
circuitBreaker.withCircuitBreaker(actorRef ? Message("3"))
// Closeのまま
```

2回`Failure`させてOpenにする。

```scala
// `maxFailure`を超える2回の失敗
circuitBreaker.withCircuitBreaker(actorRef ? PanicMessage)
circuitBreaker.withCircuitBreaker(actorRef ? HeavyMessage)
// `HeavyMesage`の処理が終わるまで待つ
Thread.sleep(3000)

// Openになっているためメッセージは処理されない
circuitBreaker.withCircuitBreaker(actorRef ? Message("4"))
```

`resetTimeout`以上の時間が経つとHalf-Openに移行する。

```scala
// `resetTimeout`以上待つ
Thread.sleep(3500)
```

Half-Openの状態で処理に成功するとCloseに戻る。

```scala
circuitBreaker.withCircuitBreaker(actorRef ? Message("5"))
// Closeに戻る

// 以下は成功する
circuitBreaker.withCircuitBreaker(actorRef ? Message("6"))
circuitBreaker.withCircuitBreaker(actorRef ? Message("7"))
circuitBreaker.withCircuitBreaker(actorRef ? Message("8"))
```

# 感想

CircuitBreakerが持つClose,Half-Open,Openの3つの状態と、
失敗する可能性のある外部サービスへのアクセスなどをマッピングがうまく出来ていれば、
いわゆるレジリエントなシステムになる気がする。