A brief introduction to command query separation of Responsibility (CQRS) mode for high Concurrency and high availability

In a common three-tier architecture, data is usually modified or queried through the data access layer, usually using the same entity. In some systems with simple business logic this may be fine, but as the system logic becomes more complex and users increase, this design can cause performance problems. Although you can do some read-write separation on DB, there are still some problems in business if you mix the read-write aspects together.

This paper introduces Command Query Responsibility Segregation (CQRS), which separates service modification (Command, increment, deletion, change, which modifies system state) from Query (Query, Query, Query, Does not modify system state). This makes the logic clearer and facilitates targeted optimization of different parts. This paper first introduces the problems of traditional CRUD mode, then introduces CQRS mode, and finally demonstrates how to implement CQRS mode with a simple online diary system. To talk about reads and writes, let’s first look at the traditional CRUD problem.

A CRUD mode problem

In previous management systems, commands (usually used to update data and operate on DB) and queries (usually used to map tables in DB) used entity objects in the Repository in the data access layer. These entities could be one row of data or multiple tables in SQLServer.

Add, delete, change, and query (CRUD) operations on DB are typically performed on the entity objects of the system. For example, data is obtained through the data access layer and then passed to the presentation layer through the data transfer object DTO. Or, if the user needs to update the Data, the Data is passed to the Model through the DTO object, and then written back to the database through the Data access layer. All interactions in the system are related to Data query and storage, which can be considered as data-driven, as shown in the following figure:

For some simple systems, this CRUD design approach can meet the requirements. In particular, some code generation tools and ORM can be very convenient and fast to achieve the function.

But there are some problems with the traditional CRUD approach:

• Using the same object entity for database reads and writes can be too crude. In most cases, for example, editing may only require updating individual fields, but the entire object may need to be threaded in, and some fields do not need to be updated. You may only need individual fields in the presentation layer at query time, but you need to query and return the entire entity object.
• When the same entity object is used to read and write the same data, resource competition may occur. Therefore, locks must be added when data is written. Check whether dirty read is allowed when reading data. This increases the logic and complexity of the system and has an impact on the throughput growth of the system.
• Synchronous, direct interaction with the database can affect performance and responsiveness when large volumes of data are accessed simultaneously, and may cause performance bottlenecks.
• Since the same entity object is used in read and write operations, the management of security and permissions becomes complicated.

A very important question is whether the read/write frequency ratio in the system is biased to read or to write. Just as the time complexity of searching and modifying the general data structure is different, such a problem also needs to be considered when designing the structure of the system. The solution is to separate read and write from the database, which we often use. Let the primary database handle transactional add,Delete, and change operations (Insert,Update,Delete), let the secondary database handle query operations (Select), and database replication is used to synchronize changes resulting from transactional operations to the secondary database in the cluster. This only handles read/write separation from the DB perspective, but reads and writes from the business or system are still stored together. They all use the same entity object.

Separating read and write from business is the command query responsibility separation pattern that will be described next.

What is CQRS

private int i = 0;
private int Increase(int value)
{
    i += value;
    return i;
}
Copy the code

private void IncreaseCommand(int value)
{
    i += value;
}
private int QueryValue()
{
    return i;
}
Copy the code

When can CQRS be considered

CQRS mode has some advantages:

1. Clear division of labor, can be responsible for different parts
2. Separating the responsibilities of business commands and queries improves system performance, scalability, and security. And in the evolution of the system can maintain a high degree of flexibility, can prevent the OCCURRENCE of CRUD mode, the query or modification of one side of the change, resulting in the problem of the other side.
3. Be logical and able to see which actions or operations in the system cause changes in the system state.
4. You can move from data-driven to task-driven and event-driven.

In the following scenarios, you can consider using CQRS mode: 6. When there are many operations at the business logic layer that require the same entity or object to operate. CQRS allows us to define different entities and methods for reading and writing, which can reduce or avoid conflicting changes in one area. Task-based user interaction systems that guide users through a series of complex steps and operations often require complex domain models and the entire team is already familiar with domain-driven design techniques. The write model has a heap of command operations related to business logic, input validation, and business logic validation to ensure data consistency. The read model has no business logic or validation heap and simply returns a DTO object to provide data for the viewmodel. The read model is ultimately consistent with the write model. 8. Suitable for some systems that need to optimize query performance and write performance separately, especially systems with very high read/write ratio, horizontal scaling is necessary. For example, on many systems read operations have far more requests than write operations. To accommodate this scenario, consider extending the write model separately and running it on one or a few instances. 9. For some teams, experienced developers can focus on complex domain models that use write operations, while less experienced developers can focus on user interface read models. 10. For systems that will evolve over time in the future, may include different versions of models, or business rules change frequently 11. You need to integrate with other systems, especially with Event Sourcing, so that a temporary exception from one subsystem does not affect the rest of the system.

However, CQRS may not be appropriate in the following scenarios:

1. When the domain model or business logic is simple, using CQRS can complicate the system.
2. There is no need to use CQRS if a simple, CRUD mode user interface and associated data access operations are sufficient, these are simply adding, deleting, changing and querying data.
3. It is not appropriate to use this pattern all over the system. CQRS can be useful in specific modules in an overall data management scenario. But there are places where using CQRS can add unnecessary complexity to the system.

Iv. Relationship between CQRS and Event Sourcing

Five simple realization of CQRS

CQRS mode is simple in idea, but it is complicated in implementation. It involves DDD, as well as Event Sourcing. The CQRS model is illustrated here using the Introduction to CQRS example from CodeProject. This example is a simple online Diary system, which implements the function of adding, deleting, modifying and checking logs. The overall structure is as follows:



The figure above clearly illustrates the separation of CQRS in terms of reads and writes. In terms of reads, data is read through a QueryFacade to a database, which could be ReportingDB. Write operations are complicated. An operation is sent to a CommandBus. Then a specific CommandHandler processes the request, generates the corresponding Event, persists the Eevnt, and modifies the database using the EventBus specific EevntHandler.

The sample code can be downloaded from CodeProject. The overall structure is as follows:

Consisting of three projects, Diary.CQRS contains all domains and message objects. Configuration uses an IOC called StructMap to initialize some variables to facilitate Web calls. Web is a simple MVC3 project that has code in the Controller to interact with CQRS.

Let’s look at the Query and Command implementations:

Implementation of the Query direction

public ActionResult Index()
{
    ViewBag.Model = ServiceLocator.ReportDatabase.GetItems();
    return View();
}

public ActionResult Edit(Guid id)
{
    var item = ServiceLocator.ReportDatabase.GetById(id);
    var model = new DiaryItemDto()
    {
        Description = item.Description,
        From = item.From,
        Id = item.Id,
        Title = item.Title,
        To = item.To,
        Version = item.Version
    };
    return View(model);
}
Copy the code

public class ReportDatabase : IReportDatabase
{
    static List<DiaryItemDto> items = new List<DiaryItemDto>();

    public DiaryItemDto GetById(Guid id)
    {
        return items.Where(a => a.Id == id).FirstOrDefault();
    }

    public void Add(DiaryItemDto item)
    {
        items.Add(item);
    }

    public void Delete(Guid id)
    {
        items.RemoveAll(i => i.Id == id);
    }

    public List<DiaryItemDto> GetItems()
    {
        returnitems; }}Copy the code

Implementation of the Command direction

The implementation of Command is complex, as illustrated by simply creating a new log.

In MVC Control, you can see that the Add Controller calls only one sentence:

[HttpPost]
public ActionResult Add(DiaryItemDto item)
{
    ServiceLocator.CommandBus.Send(new CreateItemCommand(Guid.NewGuid(), item.Title, item.Description, - 1, item.From, item.To));

    return RedirectToAction("Index");
}
Copy the code

We declare a CreateItemCommand, which simply holds the necessary information.

public class CreateItemCommand:Command
{
    public string Title { get; internal set; }
    public string Description { get;internal set; }
    public DateTime From { get; internal set; }
    public DateTime To { get; internal set; }

    public CreateItemCommand(Guid aggregateId, string title, 
        string description,int version,DateTime from, DateTime to)
        : base(aggregateId,version)
    {
        Title = title;
        Description = description;
        From = from; To = to; }}Copy the code

The Command is then sent to the CommandBus, which lets the CommandBus select the appropriate CommandHandler to handle.

public class CommandBus:ICommandBus
{
    private readonly ICommandHandlerFactory _commandHandlerFactory;

    public CommandBus(ICommandHandlerFactory commandHandlerFactory)
    {
        _commandHandlerFactory = commandHandlerFactory;
    }

    public void Send<T> (T command) where T : Command
    {
        var handler = _commandHandlerFactory.GetHandler<T>();
        if(handler ! =null)
        {
            handler.Execute(command);
        }
        else
        {
            throw new UnregisteredDomainCommandException("no handler registered"); }}}Copy the code

One of the things to note about this is the CommandHandlerFactory GetHandler method, which accepts a generic type of type T. Here is the CreateItemCommand we passed in earlier. Look at his GetHandler method.

public class StructureMapCommandHandlerFactory : ICommandHandlerFactory
{
    public ICommandHandler<T> GetHandler<T> () where T : Command
    {
        var handlers = GetHandlerTypes<T>().ToList();

        var cmdHandler = handlers.Select(handler => 
            (ICommandHandler<T>)ObjectFactory.GetInstance(handler)).FirstOrDefault();
            
        return cmdHandler;
    }
        
    private IEnumerable<Type> GetHandlerTypes<T> () where T : Command
    {
        var handlers = typeof(ICommandHandler<>).Assembly.GetExportedTypes()
            .Where(x => x.GetInterfaces()
                .Any(a => a.IsGenericType && a.GetGenericTypeDefinition() == typeof(ICommandHandler<>) ))
                .Where(h=>h.GetInterfaces()
                    .Any(ii=>ii.GetGenericArguments()
                        .Any(aa=>aa==typeof(T)))).ToList();

           
        returnhandlers; }}Copy the code

As you can see, it first looks for the types of all interfaces that implement ICommandHandler in the current assembly (ICommandHandler), and then looks for all types that implement the generic interface and whose type parameter is type T. Use the above code as an example to find the type that implements the ICommandHandler interface. As you can see, the CreateItemCommandHandler type.

public class CreateItemCommandHandler : ICommandHandler<CreateItemCommand>
{
    private IRepository<DiaryItem> _repository;

    public CreateItemCommandHandler(IRepository<DiaryItem> repository)
    {
        _repository = repository;
    }

    public void Execute(CreateItemCommand command)
    {
        if (command == null)
        {
            throw new ArgumentNullException("command");
        }
        if (_repository == null)
        {
            throw new InvalidOperationException("Repository is not initialized.");
        }
        var aggregate = new DiaryItem(command.Id, command.Title, command.Description, command.From, command.To);
        aggregate.Version = - 1; _repository.Save(aggregate, aggregate.Version); }}Copy the code

Once found, the object is returned using IOC instantiation.

Now in CommandBus, you have a Handler that handles that particular Command. The Execute method of this type is then executed.

You can see that a DiaryItem object named Aggregate is instantiated in this type. This is a little bit different from the DiaryItemDto that we used for our query, which is a domain object that contains a series of events.

public class DiaryItem : AggregateRoot.IHandle<ItemCreatedEvent>,
    IHandle<ItemRenamedEvent>,
    IHandle<ItemFromChangedEvent>, 
    IHandle<ItemToChangedEvent>,
    IHandle<ItemDescriptionChangedEvent>,
    IOriginator
{
    public string Title { get; set; }

    public DateTime From { get; set; }
    public DateTime To { get; set; }
    public string Description { get; set; }

    public DiaryItem(){}public DiaryItem(Guid id,string title, string description,  DateTime from, DateTime to)
    {
        ApplyChange(new ItemCreatedEvent(id, title,description, from, to));
    }

    public void ChangeTitle(string title)
    {
        ApplyChange(new ItemRenamedEvent(Id, title));
    }

    public void Handle(ItemCreatedEvent e)
    {
        Title = e.Title;
        From = e.From;
        To = e.To;
        Id = e.AggregateId;
        Description = e.Description;
        Version = e.Version;
    }

    public void Handle(ItemRenamedEvent e){ Title = e.Title; }... }Copy the code

The ItemCreatedEvent event is defined as follows and is used to store the data needed during the transfer.

public class ItemCreatedEvent:Event
{
    public string Title { get; internal set; }
    public DateTime From { get; internal set; }
    public DateTime To { get; internal set; }
    public string Description { get;internal set; }

    public ItemCreatedEvent(Guid aggregateId, string title ,
        string description, DateTime from, DateTime to)
    {
        AggregateId = aggregateId;
        Title = title;
        From = from; To = to; Description = description; }}Copy the code

In addition to defining the basic fields, the Domain object defines some events. For example, in the constructor, it actually initiates an event called ItemCreateEvent. It also defines the processing time logic, which is sent in the interface method named Handle. For example, ItemCerateEvent is handled by the Handle(ItemCreateEvent) method.

The ApplyChange method is in the AggregateRoot object, which is the aggregation root, a concept in DDD. All objects can be strung together through this root. This class implements the IEventProvider interface, which stores all uncommitted changes in _changes, where the ApplyChange method is used to find Eventhandler for a particular Event:

public abstract class AggregateRoot : IEventProvider
{
    private readonly List<Event> _changes;

    public Guid Id { get; internal set; }
    public int Version { get; internal set; }
    public int EventVersion { get; protected set; }

    protected AggregateRoot()
    {
        _changes = new List<Event>();
    }

    public IEnumerable<Event> GetUncommittedChanges()
    {
        return _changes;
    }

    public void MarkChangesAsCommitted()
    {
        _changes.Clear();
    }

    public void LoadsFromHistory(IEnumerable<Event> history)
    {
        foreach (var e in history) ApplyChange(e, false);
        Version = history.Last().Version;
        EventVersion = Version;
    }

    protected void ApplyChange(Event @event)
    {
        ApplyChange(@event, true);
    }

    private void ApplyChange(Event @event.bool isNew)
    {
        dynamic d = this;

        d.Handle(Converter.ChangeTo(@event, @event.GetType()));
        if(isNew) { _changes.Add(@event); }}}Copy the code

In the implementation of ApplyChange, this is actually the Domain object that implements the DiaryItem of the AggregateRoot, calling the Handle method which is the behavior we defined in the DiaryItem. The event is then saved in an internal list of uncommitted events. Related information and events are stored in the defined Aggregate object and returned.

Then Command continues and calls _repository.Save(aggregate, aggregate.version); This method. Look at the Repository object first.

public class Repository<T> : IRepository<T> where T : AggregateRoot.new()
{
    private readonly IEventStorage _storage;
    private static object _lockStorage = new object(a);public Repository(IEventStorage storage)
    {
        _storage = storage;
    } 

    public void Save(AggregateRoot aggregate, int expectedVersion)
    {
        if (aggregate.GetUncommittedChanges().Any())
        {
            lock (_lockStorage)
            {
                var item = new T();

                if(expectedVersion ! =- 1)
                {
                    item = GetById(aggregate.Id);
                    if(item.Version ! = expectedVersion) {throw new ConcurrencyException(string.Format("Aggregate {0} has been previously modified", item.Id)); } } _storage.Save(aggregate); }}}public T GetById(Guid id)
    {
        IEnumerable<Event> events;
        var memento = _storage.GetMemento<BaseMemento>(id);
        if(memento ! =null)
        {
            events = _storage.GetEvents(id).Where(e=>e.Version>=memento.Version);
        }
        else
        {
            events = _storage.GetEvents(id);
        }
        var obj = new T();
        if(memento! =null)
            ((IOriginator)obj).SetMemento(memento);
            
        obj.LoadsFromHistory(events);
        returnobj; }}Copy the code

This method is primarily used to persist events. All aggregation changes are stored in the Repository. First, the Repository checks whether the current aggregation is consistent with the aggregation stored in storage. If it is inconsistent, the object has been changed elsewhere, throwing a ConcurrencyException. Otherwise, save the change in the Event Storage.

IEventStorage is used to store all events and its implementation type is InMemoryEventStorage.

public class InMemoryEventStorage:IEventStorage
{
    private List<Event> _events;
    private List<BaseMemento> _mementos;

    private readonly IEventBus _eventBus;

    public InMemoryEventStorage(IEventBus eventBus)
    {
        _events = new List<Event>();
        _mementos = new List<BaseMemento>();
        _eventBus = eventBus;
    }

    public IEnumerable<Event> GetEvents(Guid aggregateId)
    {
        var events = _events.Where(p => p.AggregateId == aggregateId).Select(p => p);
        if (events.Count() == 0)
        {
            throw new AggregateNotFoundException(string.Format("Aggregate with Id: {0} was not found", aggregateId));
        }
        return events;
    }

    public void Save(AggregateRoot aggregate)
    {
        var uncommittedChanges = aggregate.GetUncommittedChanges();
        var version = aggregate.Version;
            
        foreach (var @event in uncommittedChanges)
        {
            version++;
            if (version > 2)
            {
                if (version % 3= =0)
                {
                    var originator = (IOriginator)aggregate;
                    var memento = originator.GetMemento();
                    memento.Version = version;
                    SaveMemento(memento);
                }
            }
            @event.Version=version;
            _events.Add(@event);
        }
        foreach (var @event in uncommittedChanges)
        {
            vardesEvent = Converter.ChangeTo(@event, @event.GetType()); _eventBus.Publish(desEvent); }}public T GetMemento<T> (Guid aggregateId) where T : BaseMemento
    {
        var memento = _mementos.Where(m => m.Id == aggregateId).Select(m=>m).LastOrDefault();
        if(memento ! =null)
            return (T) memento;
        return null;
    }

    public void SaveMemento(BaseMemento memento){ _mementos.Add(memento); }}Copy the code

In the GetEvent method, all of the aggregate root ID-related events are found. In the Save method, all events are saved in memory and snapshots are created every three events. You can see that memo mode is used here.

Then, in the Foreach loop, EventBus publishes the event for all uncommitted changes.

Now, all of the changes have been recorded. Events have been published to EventBus, and the corresponding EventHandler processes the corresponding events and interacts with the DB. Now look at EventBus’s Publish method.

public class EventBus:IEventBus
{
    private IEventHandlerFactory _eventHandlerFactory;

    public EventBus(IEventHandlerFactory eventHandlerFactory)
    {
        _eventHandlerFactory = eventHandlerFactory;
    }
        
    public void Publish<T> (T @event) where T : Event
    {
        var handlers = _eventHandlerFactory.GetHandlers<T>();
        foreach (var eventHandler inhandlers) { eventHandler.Handle(@event); }}}Copy the code

The Publish method in EventBus is similar to the Send method in CommandBus, in that the EventHandlerFactory is used to find the corresponding EventHandler, and then the Handler method is called. Such as

public class StructureMapEventHandlerFactory : IEventHandlerFactory
{
    public IEnumerable<IEventHandler<T>> GetHandlers<T> () where T : Event
    {
        var handlers = GetHandlerType<T>();
            
        var lstHandlers = handlers.Select(handler => (IEventHandler<T>) ObjectFactory.GetInstance(handler)).ToList();
        return lstHandlers;
    }

    private static IEnumerable<Type> GetHandlerType<T> () where T : Event
    {
           
        var handlers = typeof(IEventHandler<>).Assembly.GetExportedTypes()
            .Where(x => x.GetInterfaces()
                .Any(a => a.IsGenericType && a.GetGenericTypeDefinition() == typeof(IEventHandler<>)))
                .Where(h => h.GetInterfaces()
                    .Any(ii => ii.GetGenericArguments()
                        .Any(aa => aa == typeof(T))))
                 .ToList();
        returnhandlers; }}Copy the code

The ItemCreatedEventHandler object is then returned and instantiated as follows:

public class ItemCreatedEventHandler : IEventHandler<ItemCreatedEvent>
{
    private readonly IReportDatabase _reportDatabase;
    public ItemCreatedEventHandler(IReportDatabase reportDatabase)
    {
        _reportDatabase = reportDatabase;
    }
    public void Handle(ItemCreatedEvent handle)
    {
        DiaryItemDto item = newDiaryItemDto() { Id = handle.AggregateId, Description = handle.Description, From = handle.From, Title = handle.Title, To=handle.To, Version = handle.Version }; _reportDatabase.Add(item); }}Copy the code

You can see that in the Handler method, you get the parameters from the event, create a NEW DTO object, and then update that object to the DB.

At this point, the entire Command is complete.

Six conclusion