Introduction to CQRS Architecture

You should all be familiar with the CQRS (Command Query Responsibility Segment) architecture. Simply put, it is a system that is architecturally divided into two parts: command processing (write request) + query processing (read request). The read and write sides can then be implemented using different architectures to optimize both ends of the CQ (i.e., Command Side, C-side for short; Query Side, Q-side for short). As an architecture based on the idea of ​​read-write separation, CQRS does not impose too many constraints on data storage. Therefore, I think CQRS can be implemented at different levels, such as:

1. The databases at both ends of CQ are shared, and the two ends of CQ are only separated in the upper-level code; this approach has the advantage of separating code reading and writing, making it easier to maintain, and there is no data consistency problem at both ends of CQ because they share a database. I personally think that this architecture is very practical, as it not only takes into account the strong consistency of data, but also makes the code easier to maintain.
2. The databases and upper-level codes at both ends of CQ are separated, and then the data of Q is synchronized from the C end, generally through Domain Event. There are two ways of synchronization, synchronous or asynchronous. If strong consistency is required on both ends of CQ, synchronization is required; if the final consistency of data on both ends of CQ is acceptable, asynchronous can be used. With this kind of architecture, I personally think that it makes sense for the C end to adopt the Event Sourcing (ES for short) mode, otherwise it will be troublesome for yourself. Because by doing this, you will find that there will be redundant data. The same data exists in the db on the C end and in the db on the Q end. Compared with the first approach above, I can't think of any benefits. If ES is used, all the *** data on the C end can be expressed with Domain Event; and to query the data for display, you can query from the ReadDB (relational database) on the Q end.

I think there is a lot more to talk about in order to achieve high performance. Below I would like to focus on some of the design ideas I have in mind:

Avoid resource competition

Analysis of examples of flash sales

I think this is a very important point. What is resource contention? I think it is when multiple threads modify the same data at the same time. Just like Alibaba's flash sale, when the flash sale starts, many people grab a product at the same time, causing the product's inventory to be updated concurrently and reduced. This is an example of resource contention. Generally, if the resource competition is not fierce, it doesn't matter and will not affect performance; but if it is a scenario like flash sale, the db will not be able to withstand it. In a scenario like flash sale, a large number of threads need to update the same record at the same time, which in turn causes a large number of threads to accumulate inside MySQL, causing great damage to service performance and stability. What should I do? I remember that Ding Qi from Alibaba wrote a sharing. The idea is that when multiple threads on the MySQL server modify a record at the same time, these modification requests can be queued, and then for the InnoDB engine layer, it is serial. After queuing in this way, no matter how many parallel requests to modify the same row are sent by the upper-level application, for the MySQL Server side, the internal modification requests for the same row will always be queued for processing; this ensures that there will be no concurrency, so that threads will not be wasted and accumulated, resulting in a decrease in database performance. This solution can be shown in the figure below:

As shown in the figure above, when many requests need to modify the A record, the MySQL Server will queue these requests internally, and then submit the modification requests for A to the InnoDB engine layer one by one. This seems to be a queue, but in fact it will ensure that the MySQL Server will not die and can ensure stable TPS to the outside world.

However, there is still room for optimization in the scenario of flash sales, which is the Group Commit technology. Group Commit is to merge multiple requests into one operation for processing. During the flash sales, everyone is buying this product, A buys 2 pieces, B buys 3 pieces, and C buys 1 piece; in fact, we can merge the three requests of A, B, and C into one inventory reduction operation, that is, 6 pieces are reduced at one time. In this way, for the three requests of A, B, and C, we only need to do one inventory reduction operation at the InnoDB layer. Assuming that the size of each batch of our Group Commit is 50, that is, 50 reduction operations can be merged into one reduction operation and then submitted to InnoDB. In this way, the TPS of commodity inventory reduction in the flash sales scenario will be greatly improved. However, the size of each batch of this Group Commit is not the larger the better, but it is necessary to test it according to the concurrency and the actual situation of the server to get a best value. Through the Group Commit technology, according to Ding Qi's PPT, the TPS performance of commodity inventory reduction has been improved from the original 15,000 to 85,000.

From the above example, we can see how Alibaba optimizes MySQL Server to achieve high-concurrency inventory reduction in actual scenarios. However, most people don't know this technology! Because not many people have the ability to optimize the MySQL server, queuing is not possible, let alone Group Commit. This function is not built-in MySQL Server, but needs to be implemented by yourself. However, I think we can all learn from this idea.

How to avoid resource competition with CQRS

So how to design the CQRS architecture according to this idea? I would like to focus on the second CQRS architecture I mentioned above. For the C side, our goal is to process as many commands as possible within 1s, that is, data write requests. In the four-layer architecture of classic DDD, we have a mode called the unit of work mode, that is, the Unit of Work (UoW) mode. Through this mode, we can submit the modifications of multiple objects involved in the current request to the DB in a transactional manner at the application layer at one time. The DbContext of Microsoft's EF Entity Framework is an implementation of the UoW mode. The advantage of this approach is that a request to modify multiple aggregate roots can achieve strong consistency because it is a transaction. However, this approach, in fact, does not comply with the principle of avoiding resource competition. Imagine that transaction A wants to modify three aggregate roots a1, a2, and a3; transaction B wants to modify a2, a3, and a4; transaction C wants to modify three aggregate roots a3, a4, and a5. In this way, it is easy for us to understand that these three transactions can only be executed serially because they want to modify the same resources. For example, if both transaction A and transaction B need to modify the two aggregate roots a2 and a3, only one transaction can be executed at the same time. Similarly, the same applies to transaction B and transaction C. If the concurrency of transactions A, B, and C is very high, the database will have serious concurrency conflicts or even deadlocks. How can we avoid this resource competition? I think we can take three measures:

Let a Command always modify only one aggregate root

This approach is actually to narrow the scope of the transaction and ensure that a transaction only involves the modification of one record at a time. That is to say, only the modification of a single aggregate root is a transaction, making the aggregate root the smallest unit of strong data consistency. In this way, we can realize parallel modification in the most efficient way. But you may ask, but my request will involve the modification of multiple aggregate roots. What should I do in this case? In the CQRS architecture, there is something called Saga. Saga is a technology based on event-driven thinking to implement business processes. Through Saga, we can finally implement the modification of multiple aggregate roots in an eventual consistency manner. For business scenarios involving the modification of multiple aggregate roots at a time, it can generally always be designed as a business process, that is, it can define what to do first and what to do later. For example, taking the scenario of bank transfer as an example, if it is done according to the traditional transaction approach, it may be to open a transaction first, then let account A deduct the balance, and then let account B add the balance, and finally submit the transaction; if the balance of account A is insufficient, an exception will be thrown directly. Similarly, if account B encounters an exception when adding the balance, an exception will also be thrown. The transaction will ensure atomicity and automatic rollback. In other words, data consistency has been done by DB for us.

However, if it is a Saga design, it is not like this. We will define the entire transfer process as a business process. Then, the process will include multiple aggregate roots participating in the process and a process manager (ProcessManager, stateless) for coordinating the interaction of aggregate roots. The process manager is responsible for responding to the domain events generated by each aggregate root in the process, and then sending the corresponding command according to the event, so as to continue to drive other aggregate roots to operate.

In the example of money transfer, the aggregate roots involved are: two bank account aggregate roots, a transaction aggregate root, which is responsible for storing the current state of the process, and it also maintains the rules and constraints when the process state changes; and of course there is a process manager. When the transfer starts, we will first create a Transaction aggregate root, which will then generate a TransactionStarted event, and then the process manager responds to the event and sends a Command to let the A account aggregate root reduce the balance; after the A account operation is completed, a domain event is generated; then the process manager responds to the event and sends a Command to notify the Transaction aggregate root to confirm the A account operation; after the confirmation is completed, an event will also be generated, and then the process manager responds again, and then sends a Command to notify the B account to add the balance; the subsequent steps will not be described in detail. I think the general meaning has been expressed. In short, through such a design, we can complete the entire business process in an event-driven way. If any step in the process is abnormal, we can define a compensation mechanism in the process to implement the rollback operation. Or it doesn't matter if you don't roll back, because the Transaction aggregate root records the current state of the process, so that we can easily check the transfer transactions that have not ended normally in the future. For the specific design and code, those who are interested can take a look at the bank transfer example in the ENode source code, which contains a complete implementation.

Queueing Commands that modify the same aggregate root

Similar to the design of the flash sale above, we can queue the Commands that want to modify the same aggregate root at the same time. The only difference is that the queue here is not on the MySQL Server side, but in our own program. If we use a single server to handle all the Commands, then the queue is very easy to do. That is, as long as it is in memory, when a Command is to be processed, it is determined whether the aggregate root to be modified by the current Command has been processed by a previous Command. If so, queue it; if not, execute it directly. Then, when the previous Command of this aggregate root is executed, we can process the next Command of the aggregate root; but what if it is a cluster, that is, you have not only one server processing the Command, but ten servers, what should we do? Because at the same time, it is entirely possible that two different Commands are modifying the same aggregate root. This problem is also simple, that is, we can route the Command that wants to modify the aggregate root according to the ID of the aggregate root, and according to the hashcode of the ID of the aggregate root, and then modulo the number of servers currently processing the Command, we can determine which server the current Command will be routed to for processing. In this way, we can ensure that all commands for modifying the same aggregate root instance are routed to the same server for processing when the number of servers remains unchanged. Then, combined with the queuing design we made earlier inside a single server, we can ultimately ensure that only one thread is performing modifications to the same aggregate root at the same time.

Through the above two designs, we can ensure that all commands on the C side will not have concurrent conflicts. But there is a price to pay, that is, to accept eventual consistency. For example, the idea of ​​Saga is a design implemented on the basis of eventual consistency. Then, based on the above two points of this architectural design, I think the most important thing is to achieve: 1) The distributed message queue is reliable and cannot lose messages, otherwise the Saga process will be interrupted; 2) The message queue must have high performance and support high throughput; only in this way can the overall high performance of the entire system be achieved at high concurrency. EQueue, which I developed, is a distributed message queue designed for this goal. Friends who are interested can go and learn about it.

Idempotent processing of Command and Event

The CQRS architecture is message-driven, so we should try to avoid repeated consumption of messages. Otherwise, a message may be consumed repeatedly, resulting in inconsistent final data. For the CQRS architecture, I think the idempotent processing of messages in three links should be considered.

Idempotent processing of commands

I think this is not difficult to understand. For example, in the example of transfer, if the command to deduct the balance of account A is executed repeatedly, it will cause the money to be deducted from account A twice. Then the data will be inconsistent. Therefore, we must ensure that the Command cannot be executed repeatedly. How can we ensure this? Think about how we usually do some operations to determine duplication? There are generally two ways: 1) DB creates a unique index for a certain column, so that the value of a certain column of data will not be repeated; 2) Guarantee through the program, such as first judging whether it exists through select query before inserting, if it does not exist, insert, otherwise it is considered to be repeated; Obviously, the second design cannot guarantee absolute uniqueness in the case of concurrency. Then in the CQRS architecture, I think we can ensure that the Command will not be repeated by persisting the Command and then using CommandId as the primary key. Do we need to determine whether the Command exists before executing the Command every time? No. Because the probability of Command duplication is very low, it usually only occurs when the number of our server machines changes. For example, after adding a server, it will affect the routing of the Command, which will eventually cause a certain Command to be processed repeatedly. I don't want to expand on the details here, haha. If you have any questions, please discuss them in the reply. We can also avoid this problem to the greatest extent. For example, we can add a server in advance when the system is the least busy on a certain day, so that the situation of repeated consumption of messages can be reduced to a minimum. Naturally, the repeated execution of the Command is avoided to the maximum extent. Therefore, for this reason, we do not need to determine whether the Command has been executed every time a Command is executed. Instead, as long as the Command is executed, the Command is directly persisted. Then, because the CommandId is the primary key in the db, if there is a duplication, an exception of duplicate primary key will be thrown. As long as we catch the exception, we will know that the Command already exists, which means that the Command has been processed before, so we just need to ignore the Command (of course, it cannot be ignored directly. Due to space constraints, I will not expand on it in detail here. We can discuss it in detail later). Then, if there is no problem with persistence, it means that the Command has not been executed before, then it is OK. Here, there is another problem that cannot be ignored, that is, a Command has been executed for the first time and persisted successfully, but it has not been deleted from the message queue for some reason. Therefore, when it is executed next time, an exception may be reported in the Command Handler, so in a robust way, we must catch this exception. When an exception occurs, we must check whether the Command has been executed before. If so, we must assume that the current Command is executed correctly, and then take out the events generated by the previous Command for subsequent processing. This issue is a bit in-depth, so I will not elaborate on it for the time being. If you are interested, you can chat with me privately.

Idempotent processing of event persistence

Then, because our architecture is based on ES, corresponding domain events (Domain Event) will always be generated for commands that add or modify aggregate roots. What we have to do next is to persist the events first and then distribute them to all external event subscribers. As we all know, aggregate roots have a life cycle. During their life cycle, they will experience various events, and the occurrence of events always has a certain time sequence. Therefore, in order to clarify which event occurs first and which event occurs later, we can set a version number for each event, that is, version. The version of the first event generated by the aggregate root is 1, the second is 2, and so on. Then the aggregate root itself also has a version number to record its current version. Every time it generates the next event, it can also deduce the version number of the next event to be generated based on its own version number. For example, if the current version number of the aggregate root is 5, then the version number of the next event is 6. By designing a version number for each event, we can easily implement concurrency control when the aggregate root generates events, because an aggregate root cannot generate two events with the same version number. If this happens, it means that there must be a concurrency conflict. That is to say, the same aggregate root must have been modified by two commands at the same time. Therefore, it is also easy to implement idempotent processing of event persistence. That is, the event table in the db creates a unique index for the aggregate root ID + the current version of the aggregate root. In this way, the idempotent processing of event persistence can be ensured at the db level. In addition, for the persistence of events, we can also implement Group Commit like flash sales. That is, the events generated by the Command do not need to be persisted immediately, but can be accumulated to a certain amount first, such as 50, and then all events are Group Committed at one time. Then, after the event persistence is completed, the status of each aggregate root can be modified. If a concurrency conflict occurs during the Group Commit event (due to the duplication of the version number of the event of a certain aggregate root), it can be returned to a single persistent event. Why can we do this with confidence? Because we have basically ensured that an aggregate root will only be modified by one Command at the same time. In this way, it can be basically guaranteed that there will be no version number conflicts in these Group Commit events. So, do you think that many designs are actually one ring after another? When does Group Commit start? I think it can be triggered as long as two conditions are met: 1) It can be triggered when a certain timed period arrives. This timed period can be configured according to your own business scenario, such as triggering every 50ms; 2) The events to be committed reach a certain maximum value, that is, the maximum number of events that can be persisted in each batch, such as 50 events per batch. This BatchSize also needs to be evaluated based on the actual business scenario and the comprehensive performance test of your storage db to get a most suitable value; When can Group Commit be used? I think it is only necessary to use it when the concurrency is very high and a single persistent event encounters a performance bottleneck. Otherwise, it will reduce the real-time persistence of events. Group Commit increases the number of events persisted per unit time under high concurrency. The purpose is to reduce the number of interactions between the application and the DB, thereby reducing the number of IO. Unconsciously, I have talked about the three performance optimizations mentioned at the beginning, and try to reduce IO, haha.

Idempotent processing when consuming Events

In the CQRS architecture diagram, after the event persistence is completed, the next step is to publish these events (send them to the distributed message queue) for consumption by consumers, that is, to all Event Handlers. These Event Handlers may update the ReadDB of the Q end, send emails, or call the interface of the external system. As a framework, it should be responsible for ensuring that an event is not consumed repeatedly by a certain Event Handler as much as possible, otherwise, the Event Handler itself needs to ensure it. The idempotent processing here, the way I can think of is to use a table to store information on whether an event is processed by a certain Event Handler. Before calling the Event Handler each time, determine whether the Event Handler has been processed. If not, process it, and insert a record into this table after processing. I believe this method is easy for everyone to think of. If the framework does not do this, then the Event Handler must do idempotent processing itself. This idea is the process of select if not exist, then handle, and at last insert. You can see that this process is not as rigorous as the previous two processes, because in the case of concurrency, theoretically, there will still be repeated execution of the Event Handler. Or even if it is not concurrent, it may still happen. That is, if the event handler is executed successfully, but the last insert fails, the framework will still retry the event handler. Here, you will easily think that in order to provide this idempotent support, a complete execution of the Event Handler will take a lot of time, which will *** cause delays in data updates on the Query Side. However, the idea of ​​the CQRS architecture is that the data on the Q side is synchronized from the C side through events, so the update on the Q side itself has a certain delay. This is also the reason why the CQRS architecture requires eventual consistency.

Thinking about the performance issues of idempotent processing

Analysis of CommandStore performance bottlenecks

As we all know, in the entire CQRS architecture, the generation and processing of Command and Event are very frequent, and the amount of data is also very large. So how to ensure the high performance of these steps of idempotent processing? For the idempotent processing of Command, if the performance requirements are not very high, then we can simply use a relational DB, such as Sql Server and MySQL. To achieve idempotent processing, you only need to design the primary key as CommandId. No additional unique index is required for other things. So the performance bottleneck here is equivalent to the ***TPS of a large number of insert operations on a single table. Generally, MySQL databases and SSD hard drives should have no problem achieving 2W TPS. For this table, we basically only have write operations and no read operations. Only when the Command insert encounters a primary key conflict, then we may need to occasionally read the information of the existing Command based on the primary key. Then, if the amount of data in a single table is too large, what should we do? It is to shard the table and shard the database. This is the principle of avoiding massive data that we talked about at the beginning. I think it is to avoid big data through sharding to achieve the design of bypassing the IO bottleneck. However, once it comes to sharding, it comes down to what to shard. For the table storing Command, I think it is relatively simple. We can first do top-level routing according to the type of Command (equivalent to vertical splitting according to the business), and then for Commands of the same Command type, we can route according to the hashcode of CommandId (horizontal splitting). This can solve the performance bottleneck problem of Command stored in relational DB. In fact, we can also store it through popular key/value-based NoSQL, such as locally running leveldb, or supporting distributed ssdb, or others. The specific choice can be based on your own business scenario. In short, there are many options for Command storage.

Performance bottleneck analysis of EventStore

Through the above analysis, we know that the only thing needed for Event storage is the unique index of AggregateRootId+Version, and there are no other requirements. Then it is as easy as CommandStore. If a relational DB is also used, just use AggregateRootId+Version as the joint primary key. Then if we want to split the library and table, we can first do a first-level vertical split according to AggregateRootType, that is, store the events generated by different aggregate root types separately. Then, like Command, events generated by the same aggregate root can be split according to the hashcode of AggregateRootId, and all events of the same AggregateRootId are put together. This can not only ensure the uniqueness of AggregateRootId+Version, but also ensure the horizontal splitting of data. So that the entire EventStore can be horizontally scaled. Of course, we can also use key/value-based NoSQL for storage. In addition, when we query events, we will also determine the type of aggregate root and the ID of the aggregate root. Therefore, this is consistent with the routing mechanism and will not cause us to be unable to know which partition the event of the aggregate root currently to be queried is on.

Key considerations when designing storage

When designing the storage of commands and events, I think the main consideration should be to improve the overall throughput, rather than pursuing the performance of single-machine storage. Because if our system generates 10,000 events per second on average, that would be 864 million events per day. This is already a large amount of data. Therefore, we must shard commands and events. For example, if we design 864 tables, each table generates 1 million records per day, which is within an acceptable range. Then, once we have divided 864 tables, we will definitely distribute them on different physical databases. In this way, multiple physical databases provide storage services at the same time, which can improve the overall storage throughput. I personally prefer to use MySQL for storage, because on the one hand, MySQL is open source, and there are many mature practices for sharding libraries and tables. On the other hand, relational databases are more familiar and can be better controlled than MongoDB. For example, data expansion solutions can be made by ourselves, unlike MongoDB, which helps us to store big data, but once a problem occurs, we may not be able to control it. On the other hand, regarding RT, that is, the response time when storing a single piece of data, I think that no matter it is a relational database or NoSQL, the ultimate bottleneck is disk IO. The reason why NoSQL is so fast is nothing more than asynchronous disk flushing; and relational DB is not very fast, because it has to ensure the landing of data and a higher level of data reliability. Therefore, I think that in order to ensure that data will not be lost, we should try to improve RT as much as possible, and consider using SSD hard drives. On the other hand, I think that since we have already done sharding, the pressure on a single DB will not be too great, so the RT in a general LAN will not be delayed too much, which should be acceptable.

In-Memory Mode of Aggregate Roots

In-Memory mode is also a design to reduce network IO. It allows all aggregate roots whose life cycle has not yet ended to remain in memory. When we want to modify an aggregate root, we no longer need to obtain the aggregate root from the db, update it, and then save it to the db as in the traditional way. Instead, the aggregate root is always in memory. When the Command Handler wants to modify an aggregate root, it can directly get the aggregate root object from the memory without any serialization, deserialization or IO operations. Based on the ES mode, we do not need to save the aggregate root directly, but simply save the events generated by the aggregate root. When the server loses power and the aggregate root needs to be restored, it only needs to use Event Sourcing (ES) to restore the aggregate root to the latest state.

Those who have learned about actors should also know that an actor is also an instance in the entire cluster, and each actor has its own mailbox, which is used to store all the messages that the current actor needs to process. As long as the server is powered on, the actor will always exist in the memory. Therefore, the In-Memory mode is also one of the design ideas of the actor. For example, the LMAX architecture that made a lot of noise abroad before, which claims to be able to process 6 million orders per second on a single machine and a single core, is also completely based on the in-memory mode. However, I think the LMAX architecture is only for learning. If it is to be used on a large scale, there are still many problems to be solved. Foreigners use this architecture to process orders based on specific scenarios, and the requirements for programming (code quality) and operation and maintenance are very high. If you are interested, you can search for relevant information.

The idea of ​​in-memory architecture is good. By putting all data in memory, all persistence is done asynchronously. In other words, the data in memory is the most accurate, and the data in the db is persisted asynchronously, which means that at a certain moment, some data in memory may not have been persisted to the db. Of course, if you say that your program does not need to persist data, that is another matter. If it is asynchronous persistence, the main problem is the problem of downtime recovery. Let's take a look at how the Akka framework persists the state of Akka.

1. When multiple messages are sent to an actor at the same time, all of them will be queued in the actor's mailbox first;
2. Then the actor consumes messages from the mailbox sequentially in a single thread;
3. After consuming one, an event is generated;
4. For persistent events, akka-persistence also uses ES to persist events.
5. After persistence is completed, update the actor's status;
6. After the status is updated, the next message in the mailbox is processed;

From the above process, we can see that the Akka framework essentially implements the principle of avoiding resource competition, because each actor processes each message in its mailbox in a single thread, thus avoiding concurrency conflicts. Then we can see that the Akka framework also persists the event first and then updates the actor's status. This shows that the approach adopted by Akka is also called conservative, that is, it must first ensure that the data is landed, then update the memory, and then process the next message. The truly ideal in-memory architecture should be able to ignore persistence. When the actor finishes processing a message, it immediately modifies its own status and then processes the next message immediately. Then the persistence of the events generated by the actor is completely asynchronous; that is, there is no need to wait for the persistence event to complete before updating the actor's status and then processing the next message.

I think it is not important whether it is asynchronous persistence or not, because since everyone has to face a problem, that is, to restore the state of the actor after a crash, then persistent events are inevitable. Therefore, I also think that events do not need to be asynchronously persisted. It is completely possible to synchronize and persist the generated events first, just like the akka framework, and then update the actor's state after completion. In this way, when restoring the actor's state to the latest state after a crash, you only need to simply get all events from the db, and then get the actor's latest state through ES. Then if you are worried about the performance bottleneck of event synchronization persistence, this is always inevitable. If this is not done well, the performance of the entire system will not improve, so we can use SSD, sharding, Group Commit, NoSQL and other methods to optimize the performance of persistence. Of course, if the asynchronous persistence event method is adopted, it can indeed greatly improve the processing performance of the actor. However, to achieve this, there are still some prerequisites. For example, it is necessary to ensure that there is only one instance of an actor in the entire cluster, and there cannot be two identical actors working. Because if this happens, the two identical actors will generate events at the same time, resulting in a concurrency conflict (same event version number) when the latest event is persisted. But it is very difficult to ensure that there is only one instance of an actor, because we may dynamically add servers to the cluster, and some actors must be migrated to the new server. The migration process is also very complicated. When an actor migrates from the original server to the new server, it means that the work of the actor on the original server must be stopped first. Then the actor must be started on the new server; then the messages in the mailbox of the actor on the original server must be sent to the new actor, and then the subsequent messages that may still be sent to the original actor must also be forwarded to the new actor. Then the restart of the new actor is also very complicated, because it is necessary to ensure that the state of the actor after startup must be accurate. As we know, in this pure in-memory mode, the persistence of events is asynchronous, so there may be some events still in the message queue that have not been persisted. Therefore, when restarting the actor, we must also check whether there are any unconsumed events in the message queue. If there are, we need to wait. Otherwise, the state of the actor we restore is not accurate, so we cannot ensure that the memory data is accurate, and in-memory loses its meaning. These are all troublesome technical problems. In short, it is not easy to achieve a true in-memory architecture. Of course, if you say you can use products like data grid without distribution, then that may be feasible, but this is another architecture.

As mentioned above, the core working principle of the Akka framework, as well as some other aspects, such as Akka will ensure that there is only one instance of an actor in the cluster. This is actually the same as what is said in this article, and it is also to avoid resource competition, including its mailbox. When I designed ENode before, I didn’t know about the Akka framework. Later, after I learned it, I found that the idea is so close to that of ENode, haha. For example: 1) There is only one aggregate root instance in the cluster; 2) Commands for operations on a single aggregate root are queued; 3) ES is used for state persistence; 4) Both are based on message-driven architecture. Although the implementation methods are different, the purpose is the same.

summary

This article, starting from the CQRS+Event Sourcing architecture, combined with several points to note for achieving high performance (avoiding network overhead (IO), avoiding massive data, and avoiding resource competition), analyzes some possible designs that I think of under this architecture. In the entire architecture, when a command is processed, it generally needs to do two IOs, 1) persist the command; 2) persist the event; of course, the IO of sending and receiving messages is not counted here. The entire architecture is completely message-driven, so it is essential to have a stable, scalable, and high-performance distributed message queue middleware. EQueue is an achievement in working towards this goal. At present, the TCP communication layer of EQueue can send 1 million messages in just 3 seconds on an ordinary machine with an i7 CPU; interested students can take a look. ***, the ENode framework is implemented according to the designs mentioned in this article. Interested friends are welcome to download and communicate with me!