CQRS +イベントソーシングの概要：パート2

前回の記事では、CQRS +イベントソーシングの基本から始めました。 今回は、ESをさらに詳しく見ていくことを提案します。



前の記事で投稿した例では、イベントソーシングの魔法がIRepository抽象化と2つのIRepository.Save（）およびIRepository.GetById <>（）メソッドの背後に隠されていました。

何が起こっているかをより詳細に理解するために、Rinat AbdulinのLokad IDDDサンプルプロジェクトの例を使用して、イベントストアからユニットを保存およびロードするプロセスについて話すことにしました。 彼のアプリケーションサービスでは、追加の抽象化なしでイベントストアに直接訴えるので、すべてが非常に明確に見えます。 Application ServiceはCommandHandlerに類似していますが、1つのユニットのすべてのコマンドを処理します。 このアプローチは非常に便利で、1つのプロジェクトで切り替えました。

ApplicationService

IApplicationServiceインターフェイスは非常にシンプルです。

public interface IApplicationService { void Execute(ICommand cmd); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

Executeメソッドでは、コマンドを渡し、サービスがそれらを目的のハンドラーにリダイレクトすることを期待します。



この例のRinatにはCustomer集計が1つしかないため、サービスにはCustomerApplicationServiceも1つしかありません。 したがって、実際には、ロジックを基本クラスに転送する必要はありません。 私の意見では例のための素晴らしいソリューション。



Executeメソッドは、署名と一致するWhenメソッドオーバーロードの1つにコマンドの処理を渡します。 Executeメソッドの実装は、スピーカーを使用すると非常に簡単であり、すべてのメソッドでリフレクションを実行する必要はありません。

 public void Execute(ICommand cmd) { // pass command to a specific method named when // that can handle the command ((dynamic)this).When((dynamic)cmd); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

createコマンド-CreateCustomerから始めましょう。

 [Serializable] public sealed class CreateCustomer : ICommand { public CustomerId Id { get; set; } public string Name { get; set; } public Currency Currency { get; set; } public override string ToString() { return string.Format("Create {0} named '{1}' with {2}", Id, Name, Currency); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

実際のプロジェクトでは、ほとんどの場合UIとApplicationServiceの間にメッセージキューがありますが、たとえば、Rinatはサービスアプリケーションオブジェクトに直接コマンドを送信するように制限しました（クラスApplicationServerを参照）。

CreateCustomerコマンドがExecuteメソッドに入ると、Whenメソッドにリダイレクトされます。

 public void When(CreateCustomer c) { Update(c.Id, a => a.Create(c.Id,c.Name, c.Currency, _pricingService, DateTime.UtcNow)); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

Updateメソッドでは、集計の識別子と、集計の状態を変更するメソッドを呼び出すアクションを渡します。 一般的に、私の意見では、このコンテキストでCreateメソッドを呼び出すことは少し奇妙に見えるので、集約でCreateメソッドを作成するのではなく、別のコンストラクターを作成する方が良いです。 集計を作成しているようですが、何らかの理由で、状態を変更するメソッドとしてCreateメソッドを渡します。 デザイナーにとって、これは起こらなかったでしょう。



Updateメソッドに戻り、次のタスクを実行します。1）現在の集約のすべてのイベントをロードします。2）ロードされたイベントを使用して集約を作成し、その状態を復元します。3）集約で渡されたAction実行アクションを実行し、4）新しいイベントがあれば保存します。

 void Update(CustomerId id, Action<Customer> execute) { // Load event stream from the store EventStream stream = _eventStore.LoadEventStream(id); // create new Customer aggregate from the history Customer customer = new Customer(stream.Events); // execute delegated action execute(customer); // append resulting changes to the stream _eventStore.AppendToStream(id, stream.Version, customer.Changes); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

状態回復

前回の記事で示した例では、集約の状態は集約クラスのプライベートフィールドとして保存されていました。 スナップショットを追加する場合、これはあまり便利ではありません。 毎回何らかの形で状態を吸い出すか、リフレクションを使用する必要があります。 Rinatにははるかに便利なアプローチがあります-状態にはCustomerStateと呼ばれる別のクラスがあり、集約から投影メソッドを抽出することができ、スナップショットは例にありませんが保存とロードがはるかに簡単です。

ご覧のとおり、同じ集約のイベントのリストはコンストラクターに転送されます。これは、状態を復元するために推測するのが難しくないためです。

ユニットは、状態の回復をCustomerStateクラスに委任します。

 /// <summary> /// Aggregate state, which is separate from this class in order to ensure, /// that we modify it ONLY by passing events. /// </summary> readonly CustomerState _state; public Customer(IEnumerable<IEvent> events) { _state = new CustomerState(events); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

CustomerStateクラス全体のコードを提供し、いくつかのWhen投影メソッドを削除します。

 /// <summary> /// This is the state of the customer aggregate. /// It can be mutated only by passing events to it. /// </summary> public class CustomerState { public string Name { get; private set; } public bool Created { get; private set; } public CustomerId Id { get; private set; } public bool ConsumptionLocked { get; private set; } public bool ManualBilling { get; private set; } public Currency Currency { get; private set; } public CurrencyAmount Balance { get; private set; } public int MaxTransactionId { get; private set; } public CustomerState(IEnumerable<IEvent> events) { foreach (var e in events) { Mutate(e); } } ... public void When(CustomerCreated e) { Created = true; Name = e.Name; Id = e.Id; Currency = e.Currency; Balance = new CurrencyAmount(0, e.Currency); } public void When(CustomerRenamed e) { Name = e.Name; } public void Mutate(IEvent e) { // .NET magic to call one of the 'When' handlers with // matching signature ((dynamic) this).When((dynamic)e); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

コンストラクターで確認できるように、渡されたイベントを実行し、それらをMutateメソッドに渡します。Mutateメソッドは、それらをさらに適切な投影メソッドに引き込みます。

すべてのプロパティにはプライベートセッターメソッドがあり、対応するイベントを渡すことによってのみ外部から状態を変更できることが保証されています。



状態が復元されたので、変更することができます。 状態変更メソッドに少し戻りましょう。 CreateCustomerコマンドを理解し始めたので、ユニットのCreateメソッドも見ていきます。

 public void Create(CustomerId id, string name, Currency currency, IPricingService service, DateTime utc) { if (_state.Created) throw new InvalidOperationException("Customer was already created"); Apply(new CustomerCreated { Created = utc, Name = name, Id = id, Currency = currency }); var bonus = service.GetWelcomeBonus(currency); if (bonus.Amount > 0) AddPayment("Welcome bonus", bonus, utc); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

これは、ユニットの現在の状態にアクセスできるため、ビジネスルールを確認する場所です。 Customerは1回しか作成できないというビジネスルールがあります（ただし、技術的な制限もあります）ので、作成済みのユニットでCreateを呼び出そうとすると、アクションが削除されます。

すべてのビジネスルールが満たされている場合、CustomerCreatedイベントをApplyメソッドに渡します。 Applyメソッドは非常に単純で、イベントを状態オブジェクトに渡し、現在の変更のリストに追加するだけです。

 public readonly IList<IEvent> Changes = new List<IEvent>(); readonly CustomerState _state; void Apply(IEvent e) { // pass each event to modify current in-memory state _state.Mutate(e); // append event to change list for further persistence Changes.Add(e); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ほとんどの場合、集約を使用する操作ごとに1つのイベントが発生します。 ただし、Createメソッドの場合は、2つのイベントが発生する可能性があります。

CustomerCreateイベントをApplyメソッドに渡した後、ビジネスルールでボーナス額をゼロより大きくする必要がある場合、現在の顧客にウェルカムボーナスを追加できます。 この場合、AddPayment集計メソッドが呼び出されます。これは、チェックを集計せず、CustomerPaymentAddedイベントをトリガーするだけです。

 public void AddPayment(string name, CurrencyAmount amount, DateTime utc) { Apply(new CustomerPaymentAdded() { Id = _state.Id, Payment = amount, NewBalance = _state.Balance + amount, PaymentName = name, Transaction = _state.MaxTransactionId + 1, TimeUtc = utc }); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

次に、新しいイベントを保存し、読み取りモデルで公開する必要があります。 これが次の行だと思います。

 // append resulting changes to the stream _eventStore.AppendToStream(id, stream.Version, customer.Changes);
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

確認してください...

 public void AppendToStream(IIdentity id, long originalVersion, ICollection<IEvent> events) { if (events.Count == 0) return; var name = IdentityToString(id); var data = SerializeEvent(events.ToArray()); try { _appendOnlyStore.Append(name, data, originalVersion); } catch(AppendOnlyStoreConcurrencyException e) { // load server events var server = LoadEventStream(id, 0, int.MaxValue); // throw a real problem throw OptimisticConcurrencyException.Create(server.Version, e.ExpectedStreamVersion, id, server.Events); } // technically there should be a parallel process that queries new changes // from the event store and sends them via messages (avoiding 2PC problem). // however, for demo purposes, we'll just send them to the console from here Console.ForegroundColor = ConsoleColor.DarkGreen; foreach (var @event in events) { Console.WriteLine(" {0} r{1} Event: {2}", id,originalVersion, @event); } Console.ForegroundColor = ConsoleColor.DarkGray; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

まあ、ほとんど満足しています。 イベントはシリアル化され、追加専用ストアに保存されます（削除および変更は行いません）。 しかし、Rinatは（プレゼンテーションレベルで）読み取りモデルに送信する代わりに、単にコンソールに表示します。

ただし、たとえば、これで十分です。

これがメッセージキューでどのように機能するかを確認したい場合は、以前の記事のgithubで例を見ることができます。 それを少し変更し、Event Sourcingインフラストラクチャの一部をソリューションに導入しました。 この例では、スナップショットを追加する方法を示します。

スナップショット

スナップショットは、ユニットの状態の中間スナップショットを取得するために必要です。 これにより、ユニットのイベントのフロー全体をダウンロードするのではなく、最後のスナップショットを作成した後に発生したイベントのみをダウンロードできます。

それでは、実装を見てみましょう。

UserCommandHandlerクラスには、Rinatのメソッドと非常によく似たUpdateメソッドがありますが、私にとっては、スナップショットを保存およびロードします。 100バージョンごとにスナップショットを作成します。

 private const int SnapshotInterval = 100;
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

まず、リポジトリからスナップショットを作成し、このスナップショットを作成した後に発生したイベントのストリームを読み込みます。 次に、これらすべてをユニットの設計者に渡します。

 private void Update(string userId, Action<UserAR> updateAction) { var snapshot = _snapshotRepository.Load(userId); var startVersion = snapshot == null ? 0 : snapshot.StreamVersion + 1; var stream = _eventStore.OpenStream(userId, startVersion, int.MaxValue); var user = new UserAR(snapshot, stream); updateAction(user); var originalVersion = stream.GetVersion(); _eventStore.AppendToStream(userId, originalVersion, user.Changes); var newVersion = originalVersion + 1; if (newVersion % SnapshotInterval == 0) { _snapshotRepository.Save(new Snapshot(userId, newVersion,user.State)); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

コンストラクターでは、スナップショットから状態を取得するか、スナップショットがない場合は新しい状態を作成しようとします。 受信状態では、欠落しているイベントがすべて失われ、その結果、ユニットの現在のバージョンが取得されます。

 public UserAR(Snapshot snapshot, TransitionStream stream) { _state = snapshot != null ? (UserState) snapshot.Payload : new UserState(); foreach (var transition in stream.Read()) { foreach (var @event in transition.Events) { _state.Mutate((IEvent) @event.Data); } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

アグリゲートを操作した後、アグリゲートバージョンがスナップショットを取得する間隔の倍数であるかどうかを確認し、そうであれば、新しいスナップショットをリポジトリに保存します。 UserCommandHandlerから集計の状態を取得するには、そのStateプロパティを内部ゲッターにする必要がありました。



以上で、スナップショットができました。これにより、多くのイベントがある場合に、ユニットの状態をはるかに高速に復元できます。

フィードバック

CQRS + ESのトピックに興味がある場合は、コメントで質問してください。 ハブに別名がない場合は、Skypeで私に手紙を書くこともできます 最近、チェリャビンスクの友人が私のSkypeをノックしました。彼の質問のおかげで、次の記事のアイデアがたくさんありました。 自由に使えるようになったので、もっと頻繁に書くつもりです。