快速轻巧的CQRS和事件源解决方案_c# cqrs

public static void Run(ICommandQueue commander)
{
    var alice = Guid.NewGuid();
    commander.Send(new RegisterPerson(alice, "Alice", "O'Wonderland"));
    commander.Send(new RenamePerson(alice, "Alice", "Cooper"));
}

这样运行后，读端投影看起来很好，正如预期的那样：

public static void Run(ICommandQueue commander)
{
    var henry = Guid.NewGuid();
    commander.Send(new RegisterPerson(henry, "King", "Henry I"));
    for (int i = 1; i <= 20; i++)
        commander.Send(new RenamePerson(henry, "King", "Henry " + (i+1).ToRoman()));
}

不出所料，我们在版本20上有一个快照，并在事件编号后显示了当前状态21：

public static void Run(ICommandQueue commander)
{
    var hatter = Guid.NewGuid();
    commander.Send(new RegisterPerson(hatter, "Mad", "Hatter One"));
    for (int i = 2; i <= 8; i++)
        commander.Send(new RenamePerson(hatter, "Mad", "Hatter " + i.ToWords().Titleize()));
    commander.Send(new BoxPerson(hatter));
}

如您所见，聚合不再存在于事件存储中，并且最终快照的脱机副本（以及整个事件流）已在文件系统上进行了制作。

public void Run()
{
    var login = "jack@example.com";
    var password = "Let_Me_In!";
 
    if (RegisterUser(Guid.NewGuid(), login, password)) // succeeds.
        System.Console.WriteLine($"User registration for {login} succeeded");
    
    if (!RegisterUser(Guid.NewGuid(), login, password)) // fails; duplicate login.
        System.Console.WriteLine($"User registration for {login} failed");
}
 
private bool RegisterUser(Guid id, string login, string password)
{
    bool isComplete(Guid user) { return _querySearch.IsUserRegistrationCompleted(user); }
    const int waitTime = 200; // ms
    const int maximumRetries = 15; // 15 retries (~3 seconds)
 
    _commander.Send(new StartUserRegistration(id, login, password));
 
    for (var retry = 0; retry < maximumRetries && !isComplete(id); retry++)
        Thread.Sleep(waitTime);
 
    if (isComplete(id))
    {
        var summary = _querySearch.SelectUserSummary(id);
        return summary?.UserRegistrationStatus == "Succeeded";
    }
    else
    {
        var error = $"Registration for {login} has not completed after 
                    {waitTime * maximumRetries} ms";
        throw new IncompleteUserRegistrationException(error);
    }
}

请注意，上面示例中的调用方未假定命令StartUserRegistration的同步处理。而是轮询注册的状态，等待注册完成。

private bool RegisterUserNoWait(Guid id, string login, string password)
{
    bool isComplete(Guid user) { return _querySearch.IsUserRegistrationCompleted(user); }
 
    _commander.Send(new StartUserRegistration(id, login, password));
 
    Debug.Assert(isComplete(id));
 
    return _querySearch.SelectUserSummary(id).UserRegistrationStatus == "Succeeded";
}

流程管理器本身的代码比您可能想象的要简单：

public class UserRegistrationProcessManager
{
    private readonly ICommandQueue _commander;
    private readonly IQuerySearch _querySearch;
 
    public UserRegistrationProcessManager
       (ICommandQueue commander, IEventQueue publisher, IQuerySearch querySearch)
    {
        _commander = commander;
        _querySearch = querySearch;
 
        publisher.Subscribe<UserRegistrationStarted>(Handle);
        publisher.Subscribe<UserRegistrationSucceeded>(Handle);
        publisher.Subscribe<UserRegistrationFailed>(Handle);
    }
 
    public void Handle(UserRegistrationStarted e)
    {
        // Registration succeeds only if no other user has the same login name.
        var status = _querySearch
            .UserExists(u => u.LoginName == e.Name 
			    && u.UserIdentifier != e.AggregateIdentifier)
            ? "Failed" : "Succeeded";
 
        _commander.Send(new CompleteUserRegistration(e.AggregateIdentifier, status));
    }
 
    public void Handle(UserRegistrationSucceeded e) { }
 
    public void Handle(UserRegistrationFailed e) { }
}

那里有一个基本的反应式saga，它标记了使用重复用户名创建的不活动帐户。有很多的欣喜。

public static void Run(ICommandQueue commander)
{
    var alice = Guid.NewGuid();
    var tomorrow = DateTimeOffset.UtcNow.AddDays(1);
    commander.Schedule(new RegisterPerson(alice, "Alice", "O'Wonderland"), tomorrow);
 
    // After the above timer elapses, any call to Ping() executes the scheduled command.
    // commander.Ping();
}

注意，这不会在事件日志中创建任何聚合，并且命令日志现在包含计划的条目：

public void Run()
{
    // Start one account with $100.
    var bill = Guid.NewGuid();
    CreatePerson(bill, "Bill", "Esquire");
    var blue = Guid.NewGuid();
    StartAccount(bill, blue, "Bill's Blue Account", 100);
 
    // Start another account with $100.
    var ted = Guid.NewGuid();
    CreatePerson(ted, "Ted", "Logan");
    var red = Guid.NewGuid();
    StartAccount(ted, red, "Ted's Red Account", 100);
 
    // Create a money transfer for Bill giving money to Ted.
    var tx = Guid.NewGuid();
    _commander.Send(new StartTransfer(tx, blue, red, 69));
}
 
private void StartAccount(Guid person, Guid account, string code, decimal deposit)
{
    _commander.Send(new OpenAccount(account, person, code));
    _commander.Send(new DepositMoney(account, deposit));
}
 
private void CreatePerson(Guid person, string first, string last)
{
    _commander.Send(new RegisterPerson(person, first, last));
}

执行该测试后，Bill的蓝色帐户余额为31美元，Ted的红色帐户余额为169美元，这与预期的一样：

public class TransferProcessManager
{
    private readonly ICommandQueue _commander;
    private readonly IEventRepository _repository;
 
    public TransferProcessManager
    (ICommandQueue commander, IEventQueue publisher, IEventRepository repository)
    {
        _commander = commander;
        _repository = repository;
 
        publisher.Subscribe<TransferStarted>(Handle);
        publisher.Subscribe<MoneyDeposited>(Handle);
        publisher.Subscribe<MoneyWithdrawn>(Handle);
    }
 
    public void Handle(TransferStarted e)
    {
        var withdrawal = new WithdrawMoney(e.FromAccount, e.Amount, e.AggregateIdentifier);
        _commander.Send(withdrawal);
    }
 
    public void Handle(MoneyWithdrawn e)
    {
        if (e.Transaction == Guid.Empty)
            return;
 
        var status = new UpdateTransfer(e.Transaction, "Debit Succeeded");
        _commander.Send(status);
 
        var transfer = (Transfer) _repository.Get<TransferAggregate>(e.Transaction).State;
 
        var deposit = new DepositMoney(transfer.ToAccount, e.Amount, e.Transaction);
        _commander.Send(deposit);
    }
 
    public void Handle(MoneyDeposited e)
    {
        if (e.Transaction == Guid.Empty)
            return;
 
        var status = new UpdateTransfer(e.Transaction, "Credit Succeeded");
        _commander.Send(status);
 
        var complete = new CompleteTransfer(e.Transaction);
        _commander.Send(complete);
    }
}

方案G：如何实现自定义事件处理程序

public void Run()
{
    // Start one account with $50,000.
    var ada = Guid.NewGuid();
    CreatePerson(ada, "Ada", "Wong");
    var a = Guid.NewGuid();
    StartAccount(ada, a, "Ada's Account", 50000);
 
    // Start another account with $25,000.
    var albert = Guid.NewGuid();
    CreatePerson(albert, "Albert", "Wesker");
    var b = Guid.NewGuid();
    StartAccount(albert, b, "Albert's Account", 100);
 
    // Create a money transfer for Ada giving money to Albert.
    var tx = Guid.NewGuid();
    _commander.Send(new StartTransfer(tx, a, b, 18000));
}
 
private void StartAccount(Guid person, Guid account, string code, decimal deposit)
{
    _commander.Send(new OpenAccount(account, person, code));
    _commander.Send(new DepositMoney(account, deposit));
}
 
private void CreatePerson(Guid person, string first, string last)
{
    _commander.Send(new RegisterPerson(person, first, last));
}

这是Visual Studio调试器的快照，查看流程管理器的代码，在发送电子邮件通知的行上有一个断点。请注意，弹出窗口中的消息正文是我们期望的：

public class UmbrellaProcessManager
{
    private IQuerySearch _querySearch;
 
    public UmbrellaProcessManager
      (ICommandQueue commander, IEventQueue publisher, IQuerySearch querySearch)
    {
        _querySearch = querySearch;
 
        publisher.Subscribe<TransferStarted>(Handle);
        commander.Override<RenamePerson>(Handle, Tenants.Umbrella.Identifier);
    }
 
    public void Handle(TransferStarted e) { }
 
    public void Handle(RenamePerson c)
    {
        // Do nothing. Umbrella does not permit renaming people.
        
        // Throw an exception to make the consequences even more severe 
		// for any attempt to rename a person...
        // throw new DisallowRenamePersonException();
    }
}

这是一个基本的测试运行，以证明此功能可以按预期进行：

public static class Test08
{
    public static void Run(ICommandQueue commander)
    {
        ProgramSettings.CurrentTenant = Tenants.Umbrella;
 
        var alice = Guid.NewGuid();
        commander.Send(new RegisterPerson(alice, "Alice", "Abernathy"));
        commander.Send(new RenamePerson(alice, "Alice", "Parks"));
    }
}

请注意，日志中只有一个事件，并且此人的姓名没有变化：

public class RenamePerson : Command
{
    public string FirstName { get; set; }
    public string LastName { get; set; }
 
    public RenamePerson(Guid id, string firstName, string lastName)
    {
        AggregateIdentifier = id;
        FirstName = firstName;
        LastName = lastName;
    }
}

注意：与“数据传输对象”相比，我更喜欢“数据包”一词，并且我知道许多读者会反对，因此请选择适合您和您的团队的术语。

public class PersonCommandSubscriber
{
    private readonly IEventRepository _repository;
    private readonly IEventQueue _publisher;
 
    public PersonCommandSubscriber
      (ICommandQueue commander, IEventQueue publisher, IEventRepository repository)
    {
        _repository = repository;
        _publisher = publisher;
 
        commander.Subscribe<RegisterPerson>(Handle);
        commander.Subscribe<RenamePerson>(Handle);
    }
 
    private void Commit(PersonAggregate aggregate)
    {
        var changes = _repository.Save(aggregate);
        foreach (var change in changes)
            _publisher.Publish(change);
    }
 
    public void Handle(RegisterPerson c)
    {
        var aggregate = new PersonAggregate { AggregateIdentifier = c.AggregateIdentifier };
        aggregate.RegisterPerson(c.FirstName, c.LastName, DateTimeOffset.UtcNow);
        Commit(aggregate);
    }
 
    public void Handle(RenamePerson c)
    {
        var aggregate = _repository.Get<PersonAggregate>(c.AggregateIdentifier);
        aggregate.RenamePerson(c.FirstName, c.LastName);
        Commit(aggregate);
    }
}

读端(Read Side)

public class PersonSummary
{
    public Guid TenantIdentifier { get; set; }
 
    public Guid PersonIdentifier { get; set; }
    public string PersonName { get; set; }
    public DateTimeOffset PersonRegistered { get; set; }
 
    public int OpenAccountCount { get; set; }
    public decimal TotalAccountBalance { get; set; }
}

事件处理程序方法的注册在事件订阅者类的构造函数中也很明显：

public class PersonEventSubscriber
{
    private readonly IQueryStore _store;
 
    public PersonEventSubscriber(IEventQueue queue, IQueryStore store)
    {
        _store = store;
 
        queue.Subscribe<PersonRegistered>(Handle);
        queue.Subscribe<PersonRenamed>(Handle);
    }
 
    public void Handle(PersonRegistered c)
    {
        _store.InsertPerson(c.IdentityTenant, c.AggregateIdentifier, 
                            c.FirstName + " " + c.LastName, c.Registered);
    }
 
    public void Handle(PersonRenamed c)
    {
        _store.UpdatePersonName(c.AggregateIdentifier, c.FirstName + " " + c.LastName);
    }
}

领域

public class PersonAggregate : AggregateRoot
{
    public override AggregateState CreateState() => new Person();
 
    public void RegisterPerson(string firstName, string lastName, DateTimeOffset registered)
    {
        // 1. Validate command
        // Omitted for the sake of brevity.
 
        // 2. Validate domain.
        // Omitted for the sake of brevity.
 
        // 3. Apply change to aggregate state.
        var e = new PersonRegistered(firstName, lastName, registered);
        Apply(e);
    }
 
    public void RenamePerson(string firstName, string lastName)
    {
        var e = new PersonRenamed(firstName, lastName);
        Apply(e);
    }
}

注意，聚合状态是在与聚合根分开的类中实现的。

public class Person : AggregateState
{
    public string FirstName { get; set; }
    public string LastName { get; set; }
    public DateTimeOffset Registered { get; set; }
 
    public void When(PersonRegistered @event)
    {
        FirstName = @event.FirstName;
        LastName = @event.LastName;
        Registered = @event.Registered;
    }
 
    public void When(PersonRenamed @event)
    {
        FirstName = @event.FirstName;
        LastName = @event.LastName;
    }
}

事件（如命令和查询）是轻量级的POCO类：

public class PersonRenamed : Event
{
    public string FirstName { get; set; }
    public string LastName { get; set; }
 
    public PersonRenamed(string first, string last) { FirstName = first; LastName = last; }
}

持久性

/// <summary>
/// Defines the base class for all commands.
/// </summary>
/// <remarks>
/// A command is a request to change the domain. It is always are named with a verb in 
/// the imperative mood, such as Confirm Order. Unlike an event, a command is not a 
/// statement of fact; it is only a request, and thus may be refused. Commands are
/// immutable because their expected usage is to be sent directly to the domain model for 
/// processing. They do not need to change during their projected lifetime.
/// </remarks>
public class Command : ICommand
{
    public Guid AggregateIdentifier { get; set; }
    public int? ExpectedVersion { get; set; }
 
    public Guid IdentityTenant { get; set; }
    public Guid IdentityUser { get; set; }
 
    public Guid CommandIdentifier { get; set; }
    public Command() { CommandIdentifier = Guid.NewGuid(); }
}

CommandQueue实现了ICommandQueue接口，该接口定义了一组用于注册订阅者和覆盖以及发送和调度命令的方法。您可以将其视为命令的服务总线。

/// <summary>
/// Represents the state (data) of an aggregate. A derived class should be a POCO
/// (DTO/Packet) that includes a When method for each event type that changes its
/// property values. Ideally, the property values for an instance of  this class 
/// should be modified only through its When methods.
/// </summary>
public abstract class AggregateState
{
    public void Apply(IEvent @event)
    {
        var when = GetType().GetMethod("When", new[] { @event.GetType() });
 
        if (when == null)
            throw new MethodNotFoundException(GetType(), "When", @event.GetType());
 
        when.Invoke(this, new object[] { @event });
    }
}

快照