layout: true .footer[ .copyright[© 2019 TIS Inc.] ] --- class: center, middle .glass[ # Akka を用いた<br>リアクティブシステムの<br>設計パターン ] --- ## 私は何者？ - *根来 和輝* .small[Negoro Kazuki] - *TIS株式会社* - テクノロジー開発部 - *ミッション* - 高可用システムを安価なインフラで - *課外活動* - Akka 実践バイブル執筆 .with-twitter-icon[[@negokaz](https://twitter.com/negokaz)] .with-github-icon[[negokaz](https://github.com/negokaz)] .float-bottom-5.right-3[ .height-8[] ] ??? 高可用システムを安価なインフラで実現するという目標を掲げて Akka や Keepalived といった技術の検証を行っています。 --- ## 今日話したいこと * リアクティブシステムの設計パターン * 特に Akka が得意なパターンをピックアップ ??? 今日はリアクティブシステムを実現するための設計パターンをいくつかご紹介したいと思います。 リアクティブシステム自体はシステムを形容する言葉なので、その実現手段は様々あります。 なので今日は特に Akka が得意とするパターンをピックアップしてご紹介します。 --- ## Akka のおさらい **並行**・**分散**アプリケーションのための**ツールキット** - アクターモデルを用いたプログラミング - クラスタリング - ストリーム処理 - テストキット - etc ... .float-bottom-6.right-5.height-7[] --- ## リアクティブシステムのおさらい **メッセージ駆動**のアーキテクチャを基礎とし、**迅速かつ一貫した応答時間**でサービスが提供できるシステム。**障害**や**負荷の変動**に直面しても即応性を維持できる。 .center[ .float-bottom-15.left-2.width-25[] ] .footnote[ 出典：リアクティブ宣言 https://www.reactivemanifesto.org/ja ] ??? はじめにリアクティブシステムのおさらいをしておきましょう。 --- ## リアクティブシステムの設計パターン 本日ご紹介するもの - 非同期/ノンブロッキング - Circuit Breaker - Event Sourcing + CQRS - Sharding ??? とは言え、もう少し具体的にどういうアーキテクチャにすれば良いのかということが気になると思います。 今日はリアクティブシステムを実現するための設計パターンのうち、この4つのパターンをご紹介します。 --- class: center, middle ## 非同期/ノンブロッキング ??? １つ目は非同期/ノンブロッキングです。 --- ## なぜ必要か？ * レイテンシを小さくする * リソースを有効活用する * 外部システムの障害時に影響を受けにくくする ??? なぜ「非同期/ノンブロッキング」というパターンがリアクティブシステムにとって必要なのでしょうか？ このパターンを使うと（リスト）のような効果が期待できます。 レイテンシを小さくし、リソースを有効活用できるようにすることで即応性を向上させることができます。 外部システムの障害時に影響を受けにくくすることで耐障害性を向上させることができます。 --- ## 同期的にAPIを呼び出す * 応答に100msかかるAPIを5つ呼び出すと... .center[ .width-20[] ] ??? 詳しく見ていきましょう。 とある API を実装することを考えてみます。 この API の中では 5 つの外部 API を同期的に呼び出しています。 それぞれの API は 100ms で応答を返します。 このとき、全ての API を呼び出すのに合計で 500ms かかることになります。 --- ## 非同期にAPIを呼び出す * 応答に100msかかるAPIを5つ呼び出すと... .center[ .width-20[] ] ??? では、外部 API の呼び出しを非同期にするとどうなるでしょうか。 それぞれの外部 API に同時にリクエストを投げて、100ms 後にそれぞれの API から応答を受け取ります。 -- .float-bottom-5[ .with-arrow.glass[1/5 の時間で応答を返せる！] ] ??? *つまり、1/5 の時間で応答を返せるようになります。* --- ## ブロックしてタスクを実行する * 待っている間もスレッドを専有する .center[ .width-20[] ] ??? 次に、スレッドの動きを見ていきましょう。 System A から System B にリクエストを送ることを考えます。 System A は同期的に System B を同期的に呼び出しています。 System B を呼び出した後、そのスレッドはレスポンスを待ち続けます。 その間、そのスレッドでは他の処理ができません。 つまり、スレッドがブロックしている状態になります。 --- ## ノンブロッキングでタスクを実行する * 待っている間は別のタスクを処理できる .center[ .width-20[] ] ??? 非同期ノンブロッキングで処理をする場合はどうなるでしょうか。 この場合は System B を呼び出したスレッドはリクエストに対するレスポンスを待たなくなります。 つまり、レスポンスが返ってくるまでの間、他の処理ができます。 これにより、より効率的にリソースを使えるようになります。 --- ## ブロックしてタスクを実行する * 相手が応答しないとき大量のスレッドを消費する .center[ .width-20[] ] ??? ノンブロッキングの利点はそれだけではありません。 何らかの障害が起きて System B からの応答が受け取れなかったときのことを考えてみます。 スレッドがブロックするようになっていると、応答が返ってこないのでずっとブロックすることになります。 そうなると、新たに System B を呼び出す処理が呼ばれると、他のスレッドを使って呼び出す必要がでてきます。 --- ## ブロックしてタスクを実行する * 相手が応答しないとき大量のスレッドを消費する .center[ .width-20[] ] ??? こうやって、たくさん System B を呼び出す処理が呼ばれたとすると、このようにスレッドがどんどん消費されていきます。 最終的にはメモリを食い潰してそのホストのパフォーマンスを劣化させます。 --- ## ノンブロッキングでタスクを実行する * 相手が応答しなくてもスレッドの消費を抑えられる .center[ .width-20[] ] ??? ノンブロッキングにしておくと、相手が応答しなかったとしても、スレッドがブロックしないので他の処理を実行でき、スレッドを大量に消費するといったことは起きません。 --- ## 非同期/ノンブロッキング の問題点 * 同期/ブロッキング ```scala val a: Response = systemA.requestSync() val b: Response = systemB.requestSync() val responseTotalSize: Int = a.size + b.size ``` ??? ただ、非同期/ノンブロッキングに処理を行う場合は少し問題点があります。 ここには同期的に処理するコードと非同期で処理するコードの例を書きました。 両者では全く同じことをしています。 --- ## 非同期/ノンブロッキング の問題点 * 非同期/ノンブロッキング ```scala val a: Future[Response] = systemA.requestAsync(); val b: Future[Response] = systemB.requestAsync(); val responseTotalSize: Future[Int] = a.flatMap { resA => b.map(resB => resA.size + resB.size) } ``` ??? これを見てわかるとおり、非同期に処理するコードは同期的に処理するコードと比べて若干複雑になります。 また、複数の非同期処理からある変数を更新する場合など、スレッドセーフになっているか注意して実装する必要があります。 --- ## 非同期/ノンブロッキング の問題点 * 非同期プログラミングに慣れるのに学習コストが必要 * スレッドセーフな実装になるよう注意する必要がある --- ## Akka のメッセージパッシング Akka の**メッセージパッシング**は非同期/ノンブロッキング ```scala actor ! Message // 応答を待たずに次の処理へ ``` **メッセージ** で - リクエスト（≒ メソッド呼び出し） - レスポンス（≒ 戻り値） を表現する ??? Akka ではメッセージパッシングというスタイルでコンポーネントの機能の呼び出しを行います。 Akka のメッセージパッシングはそれそのものが非同期/ノンブロッキングなので、 非同期/ノンブロッキングを簡単に達成できます。 --- ## Akka のメッセージパッシング 非同期処理であってもスレッドを（ほぼ）気にせず書ける ```scala // レスポンスをメッセージに変換 systemA.requestAsync() pipeTo self systemB.requestAsync() pipeTo self var responseTotalSize: Int = 0 def receive = { case resp: Response => responseTotalSize += resp.size ``` ??? `receive` メソッドはシングルスレッドで実行される。 ただ、メッセージ単位で使われるスレッドが入れ替わるため、ThreadLocal を使ったりスレッドに強く依存するライブラリを使ったりする場合は注意が必要 -- .float-bottom-10.right-6.glass-deep[ ※ 個々のアクターは**非同期**で動く ] ??? *この状態でどこが非同期になるかというと、個々のアクターが非同期で動きます* --- ## つまり 学習コストは必要なものの、非同期処理特有の - 変数を更新するときに**更新が競合しないか**？ - 値を参照するときに**古い値が見えないか**？ といったことに頭を抱えなくてよくなる --- ## 補足 メッセージパッシング（メッセージ駆動）には<br> `Future` のようなイベント駆動と比較して - スケールしやすい - リクエストの送信元から障害を隔離しやすい といったメリットがあると言われる。 そのため、リアクティブ宣言のリアクティブシステムでは<br> 「非同期/ノンブロッキング」をさらに具体化して<br> **メッセージ駆動**を基礎となる手段としている .footnote[ [用語集 - リアクティブ宣言 #メッセージ駆動（イベント駆動と対比して）](https://www.reactivemanifesto.org/ja/glossary#Message-Driven) ] ??? 少し話が逸れますが、なぜリアクティブ宣言では「メッセージ駆動」がリアクティブシステムの実現手段になっているのかという話をします。 メッセージパッシングには `Future` のようなイベント駆動型のプログラミングスタイルと比較して、スケールしやすかったり、リクエストの送信元から障害を隔離しやすいというメリットがあると言われています。 そのため、リアクティブ宣言で言われるリアクティブシステムは「非同期/ノンブロッキング」をさらに一歩進めてメッセージ駆動を基礎となる手段としています。 --- class: center, middle ## Circuit Breaker ??? 次に Circuit Breaker というパターンを見ていきます。 --- ## Circuit Breaker とは？ 外部サービスの障害時、即座にエラーの応答を返す仕組み * リトライによるネットワーク帯域の無駄遣いが減る * 障害時もレイテンシを低く抑えられる * 障害の連鎖を抑えることができる * **Close**・**Open**・**Half-Open** の状態をとる --- ## Open？ Close？ * Circuit Breaker はもともと電子工学の言葉 * 電流が流れすぎているときに電流を遮断する保護回路 .center.height-6[  ] * **Close**: 回路が**閉じた**状態 = 通信できる * **Open**: 回路が**開いた**状態 = 通信が遮断されている ??? Close と Open がややこしいので、解説しておきます (本文) --- ## Circuit Breaker のしくみ .center.height-5[  ] * **Close**: 外部サービスへリクエストする * **Open**: すぐにエラーを返す * **Half-Open**: リクエストを試す * Open から一定時間で切り替わる * リクエストが成功したら Close に戻る .footnote[ 出典：[Circuit Breaker • Akka Documentation](https://doc.akka.io/docs/akka/2.5/common/circuitbreaker.html) ] ??? システムの世界の Circuit Breaker はどのように振る舞うかというと、こうなります。 (本文) --- ## Circuit Breaker の動き **Close** のときは通信できる .center.float.bottom-10[ .height-9[] ] ??? もう少し具体的に見ていきます。 --- ## Circuit Breaker の動き 外部のサービスに障害が起きて… .center.float.bottom-10[ .height-9[] ] --- ## Circuit Breaker の動き 正しくレスポンスが受け取れなくなると **FailureCount** のカウントが始まる .center.float.bottom-10[ .height-9[] ] --- ## Circuit Breaker の動き 外部サービスへリクエストするたびに **FailureCount** が増えて .center.float.bottom-10[ .height-9[] ] --- ## Circuit Breaker の動き 増えて... .center.float.bottom-10[ .height-9[] ] --- ## Circuit Breaker の動き **FailureCount** のしきい値に達すると、**Open** になって .center.float.bottom-10[ .height-9[] ] --- ## Circuit Breaker の動き リクエストを通さなくなり、<br> **Circuit Breaker 自身がエラーの応答**を返すようになる .center.float.bottom-10[ .height-9[] ] --- ## Circuit Breaker の動き ある一定時間が経つと、**Half-Open** になり 一時的にリクエストを通すようになる .center.float.bottom-10[ .height-9[] ] --- ## Circuit Breaker の動き その時に外部サービスが正常に戻っていると .center.float.bottom-10[ .height-9[] ] --- ## Circuit Breaker の動き **Close** になり、リクエストを通すようになる .center.float.bottom-10[ .height-9[] ] --- ## 障害の連鎖を食い止める 例）マイクロサービスなシステムがあり、 復数のサービスが連携してリクエストを処理している .float.left-20.bottom-0[ .height-14[] ] ??? 「障害の連鎖を食い止める」という話を冒頭でしましたが、具体的にどういうことか見ていきます。 --- ## 障害の連鎖を食い止める あるサービスが何らかの障害で過負荷になり リクエストを処理できなくなった .float.left-20.bottom-0[ .height-14[] ] --- ## 障害の連鎖を食い止める そのサービスに依存するサービスに処理中のリクエストが 滞留し、リクエストを処理できなくなる .float.left-20.bottom-0[ .height-14[] ] --- ## 障害の連鎖を食い止める さらにそのサービスに依存するサービスにも... (ry .float.left-20.bottom-0[ .height-14[] ] --- ## 障害の連鎖を食い止める Circuit Breaker を使うことによって、この障害の連鎖を食い止めることができる .float.left-20.bottom-0[ .height-14[] ] --- ## Akka では Circuit Breaker が標準で提供されている。 - 失敗する可能性がある処理の呼び出しをラップする - **同期処理**、**非同期処理**の両方に対応 - 指数関数的にリトライ間隔を増やす<br>**exponential backoff** が設定できる .footnote[ [Circuit Breaker • Akka Documentation](https://doc.akka.io/docs/akka/2.5/common/circuitbreaker.html) ] ??? Akka では Circuit Breaker が標準で提供されています。 --- ## コード例 ```scala val breaker = new CircuitBreaker( system.scheduler, // Open に状態遷移する失敗回数の閾値 maxFailures = 5, // 処理のタイムアウト時間 callTimeout = 10.seconds, // Open から Half-Open に遷移する時間 resetTimeout = 1.minute ) // 非同期処理を行うメソッドを呼び出し breaker.withCircuitBreaker(dangerousAsyncCall) ``` ??? 使い方はこんな感じになります。 --- class: center, middle ## Event Sourcing + CQRS ??? 次に、Event Sourcing と CQRS というパターンを見ていきます。 --- ## RDBMS の弱点 RDBMS を用いたシステムはレプリケーションによって<br>データの読み込みをスケールすることはできるが<br>書き込みをスケールさせることは難しい -- **Event Sourcing** と **CQRS** というパターンを<br> 用いることによってデータの読み込みと書き込みの両方を<br> スケールできるようになる ??? アプリケーションのデータストアとしては RDBMS を用いることが多いと思いますが、そんな RDBMS には弱点があります。 --- ## Event Sourcing？ * システムの中で起きた**イベント**を永続化する * イベントをデータストアに INSERT していく ??? イベントソーシングとは、 -- .as-underlay.without-margin[ .height-13[] ] -- .as-underlay.without-margin[ .height-13[] ] -- .as-underlay.without-margin[ .height-13[] ] -- .as-underlay.without-margin[ .height-13[] ] -- .as-underlay.without-margin[ .height-13[] ] --- ## Event Sourcing のメリット **イベント**は不変（immutable）なので… * ロック不要・複数のノードに分散して書き込める * 書き込みをスケールしやすい * キャッシュ・コピー・共有が容易 * 読み込みをスケールしやすい * データの耐久性を高くできる --- ## Event Sourcing のデメリット データの集計にコストがかかる ??? ですが、Event Sourcing にはデメリットがあります。 -- .with-arrow[**CQRS** で解決できる] --- ## CQRS？ * .bold[C]ommand and .bold[Q]uery .bold[R]esponsibility .bold[S]egregation コマンドクエリ責務分離 * コマンド(書き込み)とクエリ(読み込み)を分離する .center[ .height-10[] ] ??? CQRS とは… --- ## CQRS のメリット .without-margin[ .center[ .height-8[] ] * 書き込み側と読み込み側で * 異なるDB・データ構造が使える * 別々にスケールできるようになる .with-arrow[Command-Side に Event Sourcing を使い<br>Query-Side に集計しやすい形で永続化する] ] ??? こうすることで何が嬉しいのかというと --- ## CQRS のデメリット 読み込み側への反映は非同期に行うため… * 書き込み・読み込みで**厳密な一貫性**を保てない * 十分に時間が経つと整合性がとれる**結果整合性**はある -- .with-arrow[一時的に一貫性が取れていなくても問題ないように ビジネスやシステムを設計する必要がある] --- ## とは言え… 銀行残高がマイナスにならないかをチェックするなど **最新の状態**を取得したい場合もある --- ## 最新の状態を得る方法 **Akka Persistence** を使うとESスタイルでイベントを<br>永続化しつつ（アクターの状態として）最新の状態を<br>インメモリに保持できる .as-underlay.without-margin[ .height-12[] ] --- ## コード例 ```scala var balance = 0 // 預金残高 def receive = { case cmd @ Withdraw(amount) => // 引き出し依頼 if (balance < amount) { sender() ! Error("預金残高不足") } else { persist(cmd) { cmd => balance = balance - cmd.amount sender() ! Success ``` -- .float-bottom-10.right-5.glass-deep[ このように業務処理に関わる状態を<br> 保持するアクターは **Entity** と呼ばれる ] --- ## Akka による CQRS の実装パターン .center[ .height-16[] ] ??? Akka を使って Event Sourcing + CQRS を実装した場合のアーキテクチャを俯瞰するとこのようになります。 --- ## 1台のサーバーでは捌けなくなったら… アプリケーションサーバーの負荷が高くなったので 複数ある Entity を複数のサーバーで**分散処理**させたい 同じ口座の Entity が複数のサーバーにあると、インメモリにある口座残高の**一貫性**が崩れる可能性がある Entity の状態の一貫性を保ったまま負荷分散する方法は？ ??? ですが、ここで新たな問題に直面します。 --- class: center, middle ## Sharding ??? こういうときに登場するのが Sharding です。 --- ## Sharding？ * **負荷分散**のための仕組み * 単一障害点がないため**高い可用性**を実現できる * **Entity** を複数のサーバーに分散配置し負荷分散する * 各 Entity がクラスター内で高々 1 つだけ<br>存在することを保証できる ⇒ **一貫性を保証**できる * Akka ではこれを実現するモジュールとして<br> **Cluster Sharding** という拡張機能を提供している ??? Sharding とは… --- ## Entity のおさらい * Entity（アクター）がシステムの**状態**を分散して持つ * 例えば銀行口座ごと * イベントから Entity の状態が作り出される * 状態はオンメモリで管理される * DB に保存されたイベントから状態を復旧できる * ⇒ ノードに障害があっても Entity を復旧できる * ⇒ Entity のノード間の移動も可能 --- ## Cluster Sharding の機能 .float-top-15[ * Entity は **Shard** というグループに分けられる ] .center[ .height-16[] ] ??? ここから Akka の Cluster Sharding の機能を図で説明します --- ## Cluster Sharding の機能 .float-top-15[ * Shard は **Node(サーバー)** に配置される ] .center[ .height-16[] ] --- ## Cluster Sharding の機能 .float-top-15[ * Shard は Node ごとに **ShardRegion** というグループになる ] .center[ .height-16[] ] --- ## Cluster Sharding の機能 .float-top-15[ * 各 Entity は **ID** によって識別される ] .center[ .height-16[] ] --- ## Cluster Sharding の機能 .float-top-15[ * メッセージで ID を指定しておけば、どの ShardRegion に送ってもその ID を持つ Entity に届く ] .center[ .height-16[] ] --- ## Cluster Sharding の機能 .float-top-15[ * もしも… ] .center[ .height-16[] ] --- ## Cluster Sharding の機能 .float-top-15[ * Node に障害が起きたら ] .center[ .height-16[] ] --- ## Cluster Sharding の機能 .float-top-15[ * 別のノードに Shard が移動 (復元) する ] .center[ .height-16[] ] --- ## Cluster Sharding の機能 .float-top-15[ * 新しいノードを追加すると ] .center[ .height-16[] ] --- ## Cluster Sharding の機能 .float-top-15[ * シャードが分散(リバランス)される ] .center[ .height-16[] ] --- ## まとめ - *非同期/ノンブロッキング* - 高速化、リソース効率UP、障害の影響を少なくする - *Circuit Breaker* - 障害を切り離し、障害に巻き込まれないようにする - *Event Sourcing + CQRS* - データの読込と書込の両方をスケーラブルにする - *Sharding* - 一貫性を保証しつつ負荷分散。障害からの自動回復 ??? 本日のまとめです。 -- .with-arrow[Akka は各パターンの**実現をサポート**してくれる] ??? *Akka はこれらのパターンの実現をサポートしてくれます。* --- ## And more... .height-10[] .float-top-20.right-12[ .with-checkbox-on[非同期/ノンブロッキング] .with-checkbox-off[Circuit Breaker] .with-checkbox-on[Event Sourcing + CQRS] .with-checkbox-on[Sharding] .with-checkbox-on[ストリーム処理] .with-checkbox-on[システム間連携] .with-checkbox-on[...] ] .footnote[ 出典：[Akka実践バイブル アクターモデルによる並行・分散システムの実現｜翔泳社の本](https://www.shoeisha.co.jp/book/detail/9784798153278) ] ??? もっと学びたい方はこのような本があるのでご紹介しておきます。 残念ながら 1 版には Circit Breaker の解説がありません。今日皆さんに買っていただけたら増版時に書きます！ --- class: center, middle # 質問ありますか？ .float.bottom-18.right-0[ .height-12[] ] --- ## 参考資料 - [Akka実践バイブル アクターモデルによる並行・分散システムの実現](https://www.shoeisha.co.jp/book/detail/9784798153278) - [Reactive Microservices Architecture (O’Reilly)](http://www.oreilly.com/programming/free/reactive-microservices-architecture.html) - [Reactive Design Patterns（Manning）](https://www.manning.com/books/reactive-design-patterns) ## イラスト * いらすとや: http://www.irasutoya.com/