CQRSはなぜEvent Sourcingになってしまうのか、まとめてみたいと思います。
なぜまとめるか、それはCQRSにとってEvent Sourcingはオプションだと誤解されている方が多いからです。この記事を書いてる本人も最初はそう思っていましたが、実際に開発・運用を経験してみるとCQRSにとってEvent Sourcingはほぼ必須で、認識を改めるべきだと気づきました。なので、原義に基づいたうえで、Event SourcingではないCQRSがなぜよくない設計になるのか解説します。
続きを読むDDD Community JPのほうでCQRS/Event Sourcingについて少し盛り上がったので、どういう議論をしたかまとめるのと同時に補足も追加しました。ちなみに、Event Sourcingが主題ですが、CQRSも前提として関係します。その想定で読んでいただければと。
発端はこのツイート。
これはEvent Sourcingじゃないと無理ですね。状態に基づく限り、ストリーム処理は難しいです https://t.co/prB16GJC5q
— かとじゅん (@j5ik2o) 2020年9月14日
僕が引用したツイートは松岡さんの質問箱に対するリアクションです。その質問箱に寄せられた質問は以下。
ストリームを開いてから閉じるまでのデータが変化する毎にUIで表示したい場合、DDDではどのように設計したら良いでしょうか？ DDDのリポジトリは1つのリクエストに対して1つのリクエストを返すイメージがあり、ストリームをどのような形で扱ったら良いのかつかめずにいます。
かなり抽象的な質問なのでいろいろ確認しないとわからないのですが、CQRS/Event Sourcingならどう解決するか考えてみました。こういった説明が抽象的でわかりにくいという声をよく聞くので、具体的なコード*1を交えた説明になっています。 注意事項としては、CQRS/Event Sourcingには実装コストが掛かります*2。それほどのコストを掛ける価値があるシステムなのかよく考えてくださいというのは前提としてありますので、その点は注意して読んでください。
ドメイン駆動設計を前提にAPIサーバなどを設計した場合、APIはリクエスト・レスポンスでCRUDすることが多いです。この場合、アプリケーション内部でリポジトリを使ってドメインオブジェクトを取り出すロジックになるのでリポジトリ操作もリクエスト・レスポンスに対応します。そしてCRUDは最新状態に基づく設計スタイルです。
ショッピングカート*3に商品を追加するユースケースで考えると以下のようになります。ビジネスルールを守るのがドメインオブジェクトで重要な役割を持っています。
class AddCartItemUseCase(cartRepository: CartRepository) { def execute(cartId: CartId, itemId: ItemId, num: ItemNum): Unit = { // 最新の集約(グローバルなエンティティ)をストレージから取得する val cart = cartRepository.findById(cartId) // ロジック実行: 予算超過ならカートオブジェクトが商品の追加を拒否する！ val newCart = cart.addItem(itemId, num) // 更新された最新状態をストレージに保存する cartRepository.store(newCart) } }
カート内の商品を取得するユースケースは以下のようになるでしょう。
class GetCartItemsUseCase(cartRepository: CartRepository) { def execute(userAccountId: UserAccountId): Vector[CartItem] = { cartRepository.findByUserAccountId(userAccountId).map{ cart => cart.items } } }
質問にあったように、リクエストに対してレスポンスを返すスタイルになっています。いわゆるPULL側のAPIです。REST APIであればこれで何ら問題でしょう。
今回の質問は、ストリーム接続したいということです。漠然と「ストリームで」という話ですが、扱うデータは何でしょうか。上記の質問の場合は、ドメインの最新状態をリクエスト・レスポンス型で返しますが、ストリームでは最新状態を返すのでしょうか？ほとんどケースではそうではなく、そのとき起こった出来事であるイベントをクライアントに返すことになります。
例えば、上記のAddCartItemUseCase#executeが実行されると、商品追加イベントが発生し、ストリームに接続するクライアントにそのイベントが通知されるという具合になります。サーバからクライアントへデータがPUSHされる形になり、通常はストリームの接続は永続的になります。
サーバ側のエンドポイントの実装ではイベントのサブスクライバを作ります。サブスクライバでカートイベントを受信したら、ストリームに流すだけです。以下はakka-httpでSSE(Server Sent Event)を行う場合の例です。イベントどのサーバからでも受信できるようにストレージからWrite/Readすることになると思います。
path("events" / LongNumber ) { cartId => get { complete { cartEventSubscriber.subscribe(cartId) .map(event => ServerSentEvent(event)) .keepAlive(1.second, () => ServerSentEvent.heartbeat) } } }
さて、ドメイン状態が変化したときに、イベントをパブリッシュする実装はどうしたらよいか。ユースケースから上記で説明したストレージへイベントを書き込むことになると思います。
// CRUD前提のユースケース実装 class AddCartItemUseCase(cartRepository: CartRepository) { def execute(cartId: CartId, itemId: ItemId, num: ItemNum): Unit = { val cart = cartRepository.findById(cartId) val newCart = cart.addItem(itemId, num) cartRepository.store(newCart) // (1) cartEventService.publish(CartItemAddedEvent(cartId, itemId, num)) // (2) } }
これはいわゆるPub/Subの仕組みです。で、勘違いされている方が多いのですが、CQRS/Event Sourcingとは直接関係ないです。混同しないようにしましょう。上記はイベントを送受信しているだけでCQRS/Event Sourcingではありません。
ところで、上記コードの(1)の状態更新に加え(2)のイベントの書き込みが増えたわけですが、ここで違和感を覚えます。(1)と(2)ってストレージがまず違うので同一トランザクションにできません。つまり2相コミットとかダブルコミットというやつです。不整合が起きた場合、リカバリが面倒なやつです。例えば、(1)がコミットされたあとに、(2)が失敗したらどうなるか。(1)の完了した書き込みを削除するか、(2)の書き込みが成功するまでリトライするかです。複雑なリカバリ処理を自前で実装するハメになります。
なので、ダブルコミットはできる限り避けましょうです。以下は古典ですが読むことをお勧めします。
CQRS/Event Sourcingの考案者であるGregさんもダブルコミットを推奨していないようです：(興味あればこちらも参照ください)
The two-phase commit can be expensive but for low latency systems there is a larger problem when dealing with this situation. Generally the queue itself is persistent so the event becomes written on disk twice in the two-phase commit, once to the Event Storage and once to the persistent queue. Given for most systems having dual writes is not that important but if you have low latency requirements it can become quite an expensive operation as it will also force seeks on the disk.
”二相コミットはコストがかかりますが、低レイテンシのシステムでは、この状況を扱う際にはより大きな問題があります。一般的に、キュー自体は永続的なので、イベントは二段階のコミットで二度ディスク上に書き込まれます。ほとんどのシステムでは、二重書き込みを行うことはそれほど重要ではありませんが、もし低レイテンシの要件がある場合には、 ディスク上でのシークを強制的に行うことになるため、非常に高価な操作になる可能性があります。”
The database would insure that the values of sequence number would be unique and incrementing, this can be easily done using an auto-incrementing type. Because the values are unique and incrementing a secondary process can chase the Events table, publishing the events off to their queue. The chasing process would simply have to store the value of the sequence number of the last event it had processed, it could even update this value with a two-phase commit bringing the update and the publish to the queue into the same transaction.
”データベースは、シーケンス番号の値が一意でインクリメントされることを保証しますが、これはAUTO INCREMENT型を使用して簡単に行うことができます。値が一意でインクリメントされているので、セカンダリプロセスはイベントテーブルを追いかけることができ、イベントをキューに公開することができます。追いかけるプロセスは、単に最後に処理したイベントのシーケンス番号の値を保存しておく必要があります。 この値を更新するには、二段階のコミットで更新とキューへの公開を同じトランザクションにすることもできます。”
ではどうしたらよいか。(1)と(2)は同時に二つ処理せずに、(2)のイベントだけを書き込むようにします。(1)の状態は(2)のイベントを基に別のプロセスで作り出します。これがCQRS/Event Sourcingで、すべての状態はイベントから導出できるようにします。やっと本題。
しかし、以下の(0)から(1)の処理は、最新状態に基づいているのでそのままでは使えません。設計を見直す必要があります。
def addCartItem(cartId: CartId, itemId: ItemId, num: ItemNum): Unit = { // val cart = cartRepository.findById(cartId) // (0) // val newCart = cart.addItem(itemId, num) // cartRepository.store(newCart) // (1) // イベントの書き込みはこれでよいが、↑の処理をどうしたらよいか cartEventService.publish(ItemAdded(cartId, itemId, num)) // (2) }
前述しましたが、CQRS/Event Sourcingは、簡単にいうとイベントから状態を導出するアーキテクチャです。絵で説明したほうがわかりやすいので結論となる図は以下。どーん。「え、複雑だなぁ」と思いますよね。単純なCRUDよりはそりゃ難しくなりますよ…。その代わり耐障害性やスケーラビリティが向上するのです…。一つのアプリケーションのように見えますが、コマンド側とクエリ側は分離してデプロイすることが可能です。
ジャーナルDBと呼ばれるデータベースに、カートオブジェクトが発行したカートのイベント(以下、カートイベント)が永続化されます。このイベントを使ってコマンド側にドメインオブジェクトを、クエリ側にリードモデルを構築します。(リードモデルの形式は何でもOKです。リードDBはジャーナルDBと物理的に分けるかどうかはここでは問いません。まず概念的に理解したほうがいいでしょう)。つまり状態といえるものが2種類あるわけです。リードモデルは後述しますが、まずはドメインオブジェクトから。
以下のように、永続された複数のイベントがあれば、最新のカートオブジェクトを作る(リプレイ)ことが可能です。(1)の部分でイベントを読み込み、(2)で最新のカートオブジェクトを作り出します。
def addCartItem(cartId: CartId, itemId: ItemId, num: ItemNum): Unit = { val allEvents = cartEventService.getAllEventsById(cartId) // (1) val cart = replayFunction(allEvents) // (2) val itemAdded = cart.addItem(itemId, num) cartEventService.publish(itemAdded) }
前述のコードみてこれはひどい設計だと思ったのではないでしょうか。わかります。僕も最初そう思いました…。容易に想像がつきますが、永続化されたカートイベントの件数に比例してリプレイのパフォーマンスが悪化します。なので、対策として、永続化するイベントN件ごとにカートオブジェクトの状態をスナップショットDBに保存しておき、カートオブジェクトをリプレイする際に、最新スナップショット＋それ以降に発生した差分イベントを読み込んで、高速にリプレイします。つまり、N件以上イベントがあっても、最新のスナップショットと最大N - 1件分の読み込みで済ませることができます。とはいえ、Nを小さくすると読み込む差分イベントが少なくなる一方でイベント書き込み時のスナップショットを更新する頻度も高くなります。諸刃の剣ではありますが、バランスをうまくとる必要があります。Akka(akka-persistence)でもサポートされている機能でS3やDynamoDBをスナップショットDBとして使うことができます。
以上のことを踏まえると以下のようなイメージになります。前提としてイベントやスナップショットには連番(seqNr)が振ってあるものとします。まず、最新のスナップショットとそのスナップショットのseqNr以降の差分イベントを取得します。それらを使って最新のカートオブジェクトをリプレイします。そしてオブジェクトオブジェクトの振る舞いを実行します。新しいカートオブジェクト、イベントを得ます。イベントはエンキューされますが、seqNrが1000件で割り切れるときに新しいカートオブジェクトを最新のスナップショットとして保存します。
def addCartItem(cartId: CartId, itemId: ItemId, num: ItemNum): Unit = { val snapshot = loadSnapshot(cartId) val events = cartEventService.getEventsByIdWithSeqNr(cartId, snapshot.seqNr + 1) val cart = replayFunction(snapshot, events) val (newCart, itemAdded) = cart.addItem(itemId, num) seqNr += 1 if (seqNr % 1000 == 0 ) { saveSnapshot(newCart, seqNr) } cartEventService.publish(itemAdded, seqNr) }
こうすることでリプレイ時間を短縮化できますが、CRUDのときと比較すると多くのデータを読み込んでいることがわかりますね…。もう少し効率的にならないかについては後ほど説明します。
追記:
ドメインオブジェクトはいちいちイベントからリプレイせずに、クエリ側のリードモデルをユースケースでうまく使えないの？と思うかもしれませんが、以下の問題があり難しいです。
コマンド側のドメインモデルの概要はつかめたと思いますが、クエリ側のリードモデルを考えてみましょう。
このアーキテクチャパターンによって、イベントが唯一信頼できるデータソースとなり、そのイベントを基にクエリ側のリードモデルを構築します。イベントは不変という特徴を持ちます。なので、いつでも正しいリードモデルを作り出せます。極端な話、リードモデルの設計をミスってもイベントから作り直せます。もちろん、データ更新コストはかかりますが…。
そもそもなぜコマンドとクエリでモデルを分離するのか。理由は以下の表を参照してください。つまるところ要件が違うからです。人間が閲覧するようなシステム(SoEは特に顕著)だとまずクエリ側は非正規化データを求めます。
| - | コマンド | クエリ |
|---|---|---|
| 一貫性/可用性 | トランザクション整合性を使い一貫性を重視する | 結果整合を使い可用性を重視する |
| データ構造 | トランザクション処理を行い正規化されたデータを保存することが好まれる(集約単位など) | 非正規化したデータ形式を取得することが好まれる(クライント都合のレスポンスなど) |
| スケーラビリティ | 全体のリクエスト比率とごく少数のトランザクション処理しかしない。必ずしもスケーラビリティは重要ではない | 全体のかなりのリクエスト比率を占める処理を行うため、クエリ側はスケーラビリティが重要 |
例えば、以下のような集約(グローバルなエンティティ)がある場合でも
画面は以下のように集約を二つ結合したようなデータを求めます。deptIdだけではどの部署か分からないから名前も必要なのです。*4
ドメインオブジェクトの構造にインパクトを与えるのは、このような正規と非正規の構造的な非対称性です。考えてみるとわかりますが、トランザクション処理と検索・レポートを両立するモデルの実現も理解も難しいのです。
さて、コマンドとクエリを分離しないと実装上でどんな問題が起きるか気になるところですが、以下のようなものがあります。
簡単に以下に説明します。
集約(グローバルなエンティティ)は識別のためにIDを持つため、IDから集約本体を引き当てることができます。これはある意味正引きです。
val employee = employeeRepository.findById(employeeId)
しかし、集約の他の属性(名前や他のIDなど)を使って集約(単体もしくは複数)を引き当てたい場合があります。セカンダリインデックスを使うような逆引きですね。
以下のようなコードを書いたことがありませんか？ 僕は散々書いてきました。
val employees = employeeRepository
.findByDeptIdsWithEmpNamePatterns(deptIds, empNamePatterns)
ドメインの振る舞いを起こすためというより、クライアントにクエリのレスポンスを返すためによく使っていました。これは本当にドメインの責務でしょうか。ドメインは振る舞いを起こすことが責務と捉えると、ドメインの振る舞いが伴わないこのようなリポジトリメソッドは関心が分離できていないのかもしれません。前述した GetCartItemsUseCaseクラスの実装も本当にリポジトリの責務でいいか考える必要がありそうです。本当に集約をそのまま返すならば問題ないかもしれません。次のN+1クエリを含むような問題に発展する可能性がある場合は要注意です。
APIのレスポンスでは非正規化データを返すことがほとんどです。例えば、ホテルの予約情報一覧を作るために、予約集約に関連するホテル集約、顧客集約を解決しなければならないことがあります。ここでもドメインの振る舞いは起きません。クエリだからです。これは本当にリポジトリがやるべき仕事でしょうか。僕は過去にDBAに申し訳ないと思いながらこういうN+1が大量に発生するコードを書いていました…。そもそもレスポンスとして表形式を期待するならSQLとRDBにやらせるべきではないでしょうか。
reservationRepository.findByIds(ids).map { reservation =>
val hotel = hotelRepository.findById(reservation.hotelId)
val customer = customerRepository.findById(reservation.customerId)
new ReservationDto(reservation, hotel.name, customer.name)
}
APIのレスポンスでは集約の一部だけを求めることがあります。以下は恣意的な例ですが、UIに併せて顧客名一覧を返す処理です。リポジトリで得た集約の大部分を捨てて名前だけのリストを作ります。クエリ要件を満たすために大部分のリード結果が捨てられることになります。
val customerNames = customerRepository.findByIds(ids).map { customer =>
customer.name
}
もちろん、概念モデルの粒度が大きいとI/Oコストも比例して大きくなるので、まず先に概念の大きさに目を向けるべきですが、コマンド側のドメインでクエリを頑張りすぎるとこういうことになります。
RDBにリードモデルを構築するなら、以下のように永続化されたイベントを受信して、SQLを実行するだけです(実際にはこのようなストリーム処理は単発ではなく後に発生するワークロードに備えて常時起動させる必要があります)。これはこれで簡単ですが、cartIdを指定して読み込むので、スケールしにくいです。具体的なIDを指定しないので、ある程度の並列度を維持し永続化されたイベントを全順序に読み込む必要があります。また、こういった動作をするコンシューマは障害発生時のリバランスが必要になります。僕のお勧めはKafkaを使うことです。これはこれで一本ブログ記事が書けるぐらいの知識量なので、別の機会に触れます。
cartEventSubscriber.subscribe(cartId).runWith(Sink.foreach{
case e: CartCreated => insertCartTable(e)
case e: CartItemAdded => insertCartItemTable(e)
case e: CartItemNumUpdated => updateCartItemTable(e)
case e: CartItemRemoved => deleteCartItemTable(e)
case e: CartRemoved => deleteCartTable(e)
})
ということでだいたいの雰囲気はつかめたのではないかと思います。とはいえ実装するうえではまだ課題があります。
たとえば、addCartItem内部の処理をもう少し効率化できないかの課題については、まさに分散システムの問題に直面します。
考えられる対策としては、カートオブジェクトをユースケースのスコープから外に出して、ワークロードがある間だけ起動させてキャッシュさせておく方法があります。リクエストごとに集約をリプレイしなくなるのでオーハーヘッドが軽減します。が、これは自前で実装するのは大変です。たとえば、ウェブサーバがスケールアウトして、ロードバランサから同じカートID向けの異なるリクエストが、複数サーバに飛んだこと想定してみてください。同一カートオブジェクトが別々のサーバでリプレイされ、異なるコマンドが受理されることで、同一IDなのに別々の状態が作り出されてしまいます。ある意味スプリットブレインな状態になります。
下図はイメージなので全然正確じゃないという前提ですが、前述のようなスプリットブレインにならないようにするには、集約(グローバルなエンティティ)を以下のようにシャーディングして、コマンドをルーティングするとよいわけです。つまり、同一IDの集約はクラスタ全体で1個しかないように配置すればよいでしょう…。クラスタ上で脳みそが一つしかないのでスプリットしようがないという話です。と、ここまで考えてめちゃくちゃ大変だと想像できたと思います。さらにサーバが故障した場合に別サーバに集約をテイクオーバさせるなど自前で実装したくない…。なので、AkkaやErlangなど分散システムのフレームワークなしでこういうことは辞めましょう…。AkkaではActorという軽量プロセスをクラスタ上に分散させることができます。こういう基盤なしに無茶は辞めよう…。*5
興味があれば以下参照。
正直この分野は沼感がありますが、参考になれば幸いです。
このブログが話題になってますね。制約を付けること自体はよいことだけど、無目的に適用すると害も生じると思います。 無目的という言い方はおかしいな…。外部キー制約をどのように使えばいいのか、逆にどんなときに使うとまずいのかを考えてみたいと思います。
例えば、これ。外部キー制約はできるだけ付けるとか、何も考えずに付けるとよくないと思います。
外部キー制約は、可能な限りつけるようにしています。 DBが別れている場合、外部キーはもちろん貼れないのですが、そうでない場合はとにかく何も考えず貼っています。
データベース設計の際に気をつけていること - 食べチョク開発者ブログ
いいたいことの結論はこれ。以上終了なのですが、もう少しわかりやすく書いてみよう。
何も考えずに外部キーを貼るのは良くないな。トランザクション境界の外で結果整合性を使う場合は、外部制約はつけない。つける場合は一緒に削除されるものに限定する。つまるところ更新の境界の外と内を意識してる / “データベース設計の際に気をつけていること - 食べチョ…” https://t.co/1L7sKxrlFQ
— かとじゅん (@j5ik2o) 2020年6月15日
以下のテーブルがあるとします。商品の売上を管理するテーブルです。とりあえず、何も考えずに外部キー制約をすべてに適用しています。(論理削除ではなく一旦物理削除で考えましょう。論理削除はこういった境界分析の邪魔になりますので)
ユースケースから考えて、整合性の境界を考えます。整合性には強いものと弱いものがあります。強い整合性は作成・更新・削除するときに一緒に行います。これは「トランザクション整合性」と呼ばれることがあります。弱い整合性は「結果整合性」と呼ばれることがあります。では、上記のテーブルで考えていきます。上記の図の点線はこの整合性の境界範囲を示しています。仮に全部が一つの境界内にあると考えシミュレーションします。
売上を登録するときに、一緒に個々の売上詳細を登録・更新・削除(物理削除)します。トランザクション整合性の境界(トランザクション境界)はどこからどこまでがよいでしょうか。売上には売上詳細の概念が含まれるとしているので、売上と売上詳細は同じ境界のほうが都合がよさそうです。つまり、売上と売上詳細が更新されるときは別個ではなく、不可分な一塊として扱われます。例えば、売上が先に作られて、売上詳細が後から作られることはありません。売上詳細だけが先に削除されることもなく、一緒に削除されます。
さて、売上と売上詳細は不可分な一塊です。ある売上詳細Aが存在するとき、参照する売上Bは必ず存在します。つまり売上詳細Aの売上ID(FK)は存在する売上BのIDを参照します。売上Bが存在しないとき売上詳細Aも存在しません。この外部キー制約は機能しそうですね。データを保護できそうです。
ここからが問題です。商品ID(FK)はどうなんでしょうか。ここに外部キー制約があっても本当に大丈夫でしょうか？ユースケースを見るかぎり、商品と売上(売上詳細を含む)は別々に作成・更新・削除されます。存在する売上詳細Aが削除されるとき、参照する商品Aはまだ削除されません。売上詳細Aの商品ID(FK)から商品AのIDに外部キー制約があると、売上詳細Aは削除できません。また、 ←(この表現は間違いでした。参照している側は削除できますね) 。外部キー制約の関係上、売上詳細が存在することで、商品単体での削除ができません。ユースケースを満たさなくなるので、同じトランザクション境界にできません。(え、普通は商品は削除しないだろうがという方、分かります。とりあえず最後まで読んでほしい)
なので、以下のように境界が別個になるはずですが、外部キー制約があるほうが問題になります。
「論理削除なら整合性の境界とか考える必要ないのでは？」という想定質問があるのですが、そんなことはないです。論理削除でも更新する境界がどこからどこまでかを考える必要があります。売上を消したつもりが商品も消されたら問題ですから。削除フラグが連動する範囲が整合性の境界になります。しかし、これを頭の中で整理して考えることが難しい。なので、分析時は物理削除で考えたほうが圧倒的にラクです。
ということで、この場合どう考えるとよいか。トランザクション境界が異なるということはライフサイクルの境界が異なるわけです。売上詳細から参照する商品IDは存在するかもしれないし、すでに削除されているかもしれません。こういった弱い整合性を結果整合性といいます。ここには外部キー制約は適用できません。
また、アプリケーションロジックで以下の売上合計金額を算出する場合、売上詳細からみて商品が結果整合性では、再計算できなくなってしまいます。
これに対処するには、売上・売上詳細の境界内に再計算するための材料を保持する必要があります。もしくは計算結果を保持する方法もありそうですね。前者だとしたら、更新時にそのときの商品IDの単価をコピーする必要があります。(場合によっては商品名が更新されたり削除されるかもしれないので、商品名のコピーも必要になるかもしれません)
このような考え方は、ドメイン駆動設計の集約 を学ぶとわかるようになるはずです。今回はテーブル設計の視点から説明してみましたが、ドメインモデルが整合性の境界を持つと考えるとわかりやすくなるかも。DDDの観点で簡単に言えば、強い整合性境界である集約の内側では外部キー制約に意味があり、集約の外には外部キー制約は不要です。興味がある方はぜひ学んでみたらいいと思います。
ということで、整合性の境界を無視して、外部キー制約を適用することはできません、ということで。
併せて読みたい:
追記:
ブコメで例が悪いと指摘があったので、少し再考してみた。わかりにくいことは認める。言いたいことは結果整合性を求める用途では外部キー制約は使えない。トランザクション整合性のある境界内では使えるという主張だった。
で、商品＝今使える商品という意味で捉えてほしい。商品が廃盤になるとやはり物理削除して、過去に使えてた商品テーブルに移動するかもしれない。実際は、こんな難しいことをせずに、商品は削除せずに「廃盤状態」に遷移できるようにする方法もある。が、何が正しいかは要件による。ここでは仕様の善し悪しを議論したいわけではなく、仮に商品を消すことがある場合、売上詳細から商品ID(FK)は強い整合性を望むので使えないという意図だった。つまるところ、ライフサイクルの境界が異なるのだから、商品IDから商品に到達できるかはそのとき次第なので、以下のような考慮が必要かもしれない、ということ。
商品IDだけでは困るケースがある。例えば、商品が廃盤になったり、単価改定など。過去に作った売上が変わってしまう。ライフサイクルが異なるので起きる。これを想定するなら、売上詳細側に商品名や単価などを取り込まないといけない https://t.co/tNSZJ5XgmB
— かとじゅん (@j5ik2o) 2020年6月17日
とはいえ、もっとよい事例はないかと考えた。Slackのようなチャットサービスで、あるアカウントが投稿したメッセージは他のアカウントでも読める。しかし、ある投稿者のアカウントIDが退会したケースを考えてみよう。忘れられる権利に対応するために、アカウントは物理削除しなければならないとする(物理削除好きやな。まぁ例示のための仮定です)。しかし、退会アカウントのメッセージはタイムラインで、”退会済みアカウント”が投稿したメッセージとして、他のアカウントから閲覧できるものとする。この場合でも、メッセージとアカウントはライフサイクル、整合性の境界が独立している。メッセージからみてアカウントはあるかもしれないしないかもしれない。弱い整合性。この場合は、メッセージのアカウントIDからアカウントのアカウントIDに外部キー制約は適用できない。
メッセージングシステムのお題のようです。面白そうなのでちょっと考えてみよう。
集約候補が以下の3つ。
スレッド集約はメッセージを複数保持するようです。
ということが指摘されている。まぁわかります。これはCQRS/ESなら解決できるよと言ってみる
で、僕ならどう考えて実装に落とすかつらつらまとめてみよう。CQRS/ES前提です。Akkaの成分は少なめでScalaの擬似コードで解説します。コードはコンパイルしてないので…おかしなところあるかも。
問題はスレッド集約がメッセージの集合を保持した場合、更新コストが大きくなりがち。さらに差分更新しようとしても集約の内部実装が複雑になるのではという論点。
まず考えるのは、スレッドとメッセージの関係性に強い整合性は必要か？ ということです。Noならメッセージを別の集約として切り出すのがよいとは思います。おそらく問題設定としてはYesになっている気がしますが…。 仮に契約プランによって書き込めるメッセージ数が変わるという場合は、スレッドとメッセージには強い整合性が必要かもしれません。別別の集約にした場合、トランザクションが別になり結果整合性になるため、インスタンス数(レコード数)を厳密に制御することはできませんね。ということで、強い整合性が必要という前提で進めます。
あと、この問題だけではスレッド集約がどのような使われ方をするのが厳密にはわかりません。つまり振る舞いがイメージできません。たとえば、スレッドにメンバーとして参加したアカウントだけがメッセージを閲覧・追加・更新できるというシナリオがあればスレッド集約の構造が明確になります。メンバーの考慮が必要なスレッド集約は以下のようなイメージになるでしょう。
case class Message(id: MessageId, text: MessageText, senderId: AccountId, createdAt: Instant, updatedAt: Instant) case class Messages(values: List[Message]) { def add(message: Message): Messages = ??? def update(message: Message): Messages = ??? } case class Members(values: List[AccountId]) { def add(accountId: AccountId): Members = ??? def remove(accountId: AccountId): Members = ??? def contains(accountId: AccountId): Boolean = ??? } class Thread(id: ThreadId, members: Members, messages: Messages, createdAt: Instant) { def addMember(account: AccountId): Thread = copy(members = members.add(accountId)) def addMessage(messageId: MessageId, messageText: MessageText, senderId: AccountId): Either[ThreadError, Thread] = { if (members.contains(senderId)) { Right(copy(messages = messages.add(Message(messageId, messageText, senderId, Instant.now, Instant.now)))) } else { Left(new AddMessageError) } } def updateMessage(messageId: MessageId, messageText: MessageText, sender: AccountId): Either[ThreadError, Thread] = if (members.contains(senderId)) { Right(copy(messages = messages.updateMessage(Message(messageId, messageText, senderId, Instant.now, Instant.now)))) } else { Left(new UpdateMessageError) } def getMessages(accountId: Account): Either[ThreadError, Messages] = if (members.contains(accountId)) { Right(messages) } else { Left(new ReadMessagesError) } }
課題に戻りますが、このクラスをリポジトリでCRUDすることを想像してみましょう。thread.messagesが1000件ある場合です。
for { thread <- threadRepository.findById(threadId) // 1000件メッセージを持つスレッドを取得 newThread <- thread.addMessage(Message(...)) // 状態変更 _ <- thradRepository.store(newThread) // 1001件メッセージを持つスレッドを更新 } yield ()
スレッドのどのフィールドが更新されたかわからない場合は、スレッドのタイトル、説明、メッセージの本文などをすべての情報をSQLなどを使って更新するのでしょうか？ これはあまりに非効率ではないか…。それを回避するにはThread集約内部にどのフィールドをDB上で更新すべきか把握する仕組みが必要なのではないかという話ですね。 あと、そもそもfindByIdの時点で、本文も含む1000件のメッセージをDBから取得するのは効率がよいとはいえませんね。
上記の問題をCQRS/ESで改善できるか考えてみよう。
集約はCQRSのC(コマンド)側に所属しますが、役割は副作用を起こすことを目的にします。 たとえば、メンバーの追加・削除、メッセージを追加・更新などです。メンバーの確認はメッセージの追加・更新に必要なので含めます。これら以外はクエリ側の責務とします。
class Thread(id: ThreadId, memberIds: MemberIds, messages: MessageIds, createdAt: Instant) { def addMember(account: AccountId): Thread = copy(members = memberIds.add(accountId)) def addMessage(messageId: MessageId, messageText: MessageText, senderId: AccountId): Either[ThreadError, Thread] = if (memberIds.contains(senderId)) { Right(copy(messages = messageIds.add(mesageId)) // 本文は保存しない } else { Left(new AddMessageError) } def updateMessage(messageId: MessageId, messageText: MessageText, sender: AccountId): Either[ThreadError, Thread] = if (memberIds.contains(senderId)) { Right(this) // スレッドではメッセージIDしか管理しないのでメッセージ本文を更新できない } else { Left(new UpdateMessageError) } }
本文を持たないので大分軽量化できたのではないでしょうか？(ただ、内部にID集合は保持するので無限に保持するわけには行かないとは思います。上限値は必要だと思います) さて、クエリ側にどうやって連携するかですね。メッセージの本文も読めないと意味がないですからね。
ということで、擬似コードですが、以下のように副作用が起きたときにイベントをDBに追記します。このイベントは別のプロセスがコンシュームしてリードモデルを作る元ネタになります。
sealed trait ThreadEvent case class MemeberAdded(threadId: ThreadId, accountId: AccountId, occurredAt: Instant) extends ThreadEvent case class MessageAdded(threadId: ThreadId, messageId: MessageId, messageText: MesssageText, senderId: AccountId, occurredAt: Instant) extends ThreadEvent case class MessageUpdated(threadId: ThreadId, messageId: MessageId, messageText: MesssageText, senderId: AccountId, occurredAt: Instant) extends ThreadEvent class Thread(id: ThreadId, memberIds: MemberIds, messages: MessageIds, createdAt: Instant) { def addMember(accountId: AccountId): Thread = { persistEvent(MemberAdded(id, accountId, Instant.now)) copy(members = memberIds.add(accountId)) } def addMessage(messageId: MessageId, messageText: MessageText, senderId: AccountId): Either[ThreadError, Thread] = if (memberIds.contains(senderId)) { persistEvent(MessageAdded(id, messageId, messageText, senderId, Instant.now)) Right(copy(messages = messageIds.add(mesageId)) } else { Left(new AddMessageError) } def updateMessage(messageId: MessageId, messageText: MessageText, senderId: AccountId): Either[ThreadError, Thread] = if (memberIds.contains(senderId)) { persistEvent(MessageUpdated(id, messageId, messageText, senderId, Instant.now)) Right(this) } else { Left(new UpdateMessageError) } }
書き込みはこれでよいとして、集約をDBから再生する際はどうなるのか。DBに保存してあるイベント列を順番に空のThreadに適用するだけです。 これも疑似コードで説明。この考慮はあくまでコマンドを実行するための前提を整えるためのものです。
class Thread(id: ThreadId, events: Seq[ThreadEvent]) { private var memberIds: MemberIds = MemberIds.empty private var messages: MessageIds = MessageIds.empty // イベントから最新状態を復元 events.foreach{ case MemberAdded(_, accountId, _) => members = memberIds.add(accountId) case MessageAdded(_, messageId, messageText, senderId, createdAt) => messages = messages.add(Message(messageId, messageText, senderId, createdAt) case MessageUpdated(_, messageId, messageText, senderId, updatedAt) => messages = messages.update(Message(messageId, messageText, senderId, updatedAt) } def addMember(accountId: AccountId): Thread = { persistEvent(MemberAdded(id, accountId, Instant.now)) copy(members = memberIds.add(accountId)) } def addMessage(messageId: MessageId, messageText: MessageText, senderId: AccountId): Either[ThreadError, Thread] = if (memberIds.contains(senderId)) { persistEvent(MessageAdded(id, messageId, messageText, senderId, Instant.now)) Right(copy(messages = messageIds.add(mesageId)) } else { Left(new AddMessageError) } def updateMessage(messageId: MessageId, messageText: MessageText, senderId: AccountId): Either[ThreadError, Thread] = if (memberIds.contains(senderId)) { persistEvent(MessageUpdated(id, messageId, messageText, senderId, Instant.now)) Right(this) } else { Left(new UpdateMessageError) } }
これでドメインを扱うコマンド側では、差分を表現するイベントをジャーナルとして追記保存すればいいことになります。また最新の集約の状態はコマンドを受け付けるために必要最低限のものしか持ちません。クエリをしないので、これで問題は起きません。当初想定していたモデルよりかなり小さくなります。コマンド側の集約はリプレイ後はルールをチェックしてパスしたらイベントを追記保存するだけなんです。
ここでは示してませんが、イベントの列が巨大だとリプレイに時間が掛かります。例えばイベントN件に1回スナップショットを保存しておき、リプレイ時に最新スナップショット+差分のイベントでリプレイを高速化することもできます。とはいえ、リクエストのたびにイベント列を読み込み・適用するのは効率が悪そうです。Akkaのアプローチでは軽量プロセスであるActorとしてスレッド集約を実装します。上記のクラスがActorとして実装されます(集約アクター)。メソッドコールがメッセージパッシングに変わります。すでにイベント列がある集約アクターはAkkaがストレージからイベントを読み込み、メッセージとして適用してくれます。また一定時間メッセージを受け付けていない集約アクターはランタイムから消えるように設定できます。まぁ、必要な機能が揃っているので確実に楽できます。
クエリ側のリードモデルはどうやって作るのか？ 仮に上記のイベントがDynamoDBに書き込まれるとして、パーティションキーはThreadId, ソートキーはイベント番号(akkaでは自動採番してくれます)、イベント本体はJSONなどで格納されるイメージとします。これをDynamoDB Streamsなどでコンシュームします。コンシューム部分はLambdaでもKCLでもよいです。以下を参考にしてみてください。
docs.aws.amazon.com docs.aws.amazon.com
この方法でイベントが時系列順に手に入ります。リードモデルを作るにはこのイベントを順番にリードDBに反映します。 仮にRDBにTHREADテーブル、MEMBERテーブル、MESSAGEテーブルがある場合は以下のような処理になります。 consumeEventsByThreadIdFromDDBStreamsはDynamoDB Streamsからレコードを読み込んでイベントを返す関数です。
consumeEventsByThreadIdFromDDBStreams.foreach{
case ev: ThreadCreated => insertThread(ev) // 実際にはスレッドが作られたときのイベントも永続化する必要がある
case ev: MemberAdded => insertMember(ev)
case ev: MessageAdded => insertMessage(ev)
case ev: MessageUpdated => updateMessage(ev)
}
最終的にリードモデルはレスポンスの形式に併せて定義するとよいでしょう。
case class ThreadDto(id: Long, ...) case class MemberDto(id: Long, accountId: Long, createdAt: Int) case class MessageDto(id: Long, threadId: Long, text: String, senderId: Long, createdAt: Instant, updatedAt: Instant)
リードモデルはDAO内部でSQLを使って、部分集合を取り出したりできます。リードモデルはお好きなように…。
val messages: Seq[MessageDto] = MessageDao.findAllByThreadIdWithOffsetLimit(threadId, 0, 100)
最終的な疑似コードイメージです。 1000件のメッセージを保持するスレッドといってもIDしか持ちません。本文を持つより簡単にリプレイし一つのコマンドを付けて一つのイベントを永続化するだけです。1001件のメッセージは永続化しません。間違いなくスケールするのはこっちですね。
val thread = Thread(ThreadId(1L)) // 永続化されているイベント列があれば渡される。なければイベント列は空で作られる thread.addMessage(...) // 1つのコマンドで1つのイベントが追記されるだけ。1001件も更新されません。
上記の考え方がわかったとしても、CQRS/ESは自前でやるのは大変です。 たとえば、以下の処理が複数のサーバで起こった場合、ロックがなくても大丈夫でしょうか？ダメそうですねｗ そう書き込むためのトランザクション管理が必要なのです…。
val thread = Thread(ThreadId(1L)) thread.addMessage(...)
AkkaではThread部分がActorになりますが、複数のサーバで起動しても、ユニークになるように分散できます。つまり分散システム上で同じIDのActorがたかだか一つになるように配置してくれるので、トランザクションの問題は起きません。 こういったことを考えると、CQRS/ESをサポートしたフレームワークやライブラリを使うほうが圧倒的に楽です。Akka お勧めです。
Akka実践バイブル アクターモデルによる並行・分散システムの実現
Akkaクラスターがネットワーク分断に遭遇した場合に、UnreachableメンバーをDown状態に遷移させるためのリゾルバのOSS実装を以下にまとめる。
ちなみに、このリゾルバがない場合はUnreachableのままだとリーダアクションが取れずにクラスターが機能不全状態なる。かといってAutodownを有効にするとスプリットブレインが発生する可能性がある。これを解決するのがスプリットブレインリゾルバで商用版はLightbend社から提供されている。スプリットブレインリゾルバの仕様はこちら
| git repo | stars | 備考 |
|---|---|---|
| TanUkkii007/akka-cluster-custom-downing | 131 | OldestAutoDowning, QuorumLeaderAutoDowning, MajorityLeaderAutoDowningに対応している。OldestAutoDowningには不具合があるようだ。要修正 |
| mbilski/akka-reasonable-downing | 85 | Static QuorumによるDowningにしか対応していない |
| arnohaase/simple-akka-downing | 16 | static-quorum, keep-majority, keep-oldestに対応している |
| guangwenz/akka-down-resolver | 5 | Static QuorumによるDowningにしか対応していない |
