How to avoid concurrency problems when scaling records horizontally?

Question

Suppose there is a work service that receives messages from the queue, reads the product with the specified Id from the document database, applies some message-based manipulation logic, and finally writes the updated product back to the database (a).

This work can be safely performed in parallel when working with various products, so we can scale horizontally (b). However, if more than one instance of the service works with the same product, we may encounter concurrency or concurrency exceptions from the database, in which case we must apply some retry logic (and retrying again may fail, and so on) .

Question : How do we avoid this? Is there a way to ensure that two instances do not work on the same product?

Example / Example of use . The online store has an excellent sale of ProductA, productB and productC products that end in an hour, and hundreds of buyers buy. For each purchase, a message is queued (productId, numberOfItems, price). Purpose . How can we start three instances of our work service and make sure that all messages for productA will be completed in instance A, productB up to instance B and productC with instance C (which will not lead to concurrency problems)?

Notes . My service is written in C # hosted on Azure as a worker, I am using Azure Queues for messaging, and I am thinking of using Mongo for storage. In addition, object identifiers are GUID .

It's more about technology / design, so if you use different tools to solve a problem, I'm still interested.

Answer 1

For this kind of thing, I use blob rental. Basically, I create a blob with an object identifier in some known repository account. When worker 1 takes an object, he tries to get a lease on the blob (and create a blob himself if it does not exist). If it is so successful, I allow the processing of the message. After that, always let go of the rental. If I fail, I will send the message back to the queue

I follow the description described by Steve Mark here http://blog.smarx.com/posts/managing-concurrency-in-windows-azure-with-leases , although I am configured to use new storage libraries

Edit after comments: If you have a potentially high level of messages talking to the same object (as follows from your remark), I would revise your approach somewhere .. either an entity structure or a message structure.

For example: consider the CQRS design pattern and save the changes regardless of how each message is processed. Thus, the product object now represents the totality of all changes made to the object by various employees, consistently reapplied and rehydrated into one object

Answer 2

Any solution trying to divide the load into different elements of the same collection (for example, orders) is doomed to failure. The reason is that if you get a high level of transactions, you will have to start doing one of the following:

• allow nodes to talk to each other ( hey guys, are anyone working with this? )
• Divide the formation of the identifier into segments (node ​​a creates an ID 1-1000, node B 1001-1999), etc., and then just let them process their segment
• dynamically divide the collection into segments (and each node process the segment.

so what's wrong with these approaches?

The first approach is simply copying transactions in the database. If you cannot spend a lot of time optimizing your strategy, it is best to rely on transactions.

The second two options will decrease performance, because you need to dynamically route messages by identifiers, as well as change the strategy at runtime to also include newly inserted messages. Ultimately this will fail.

Decision

Here are two solutions that you can also combine.

Retry automatically

Instead, you have an entry point somewhere that reads from a message queue.

In it, you have something like this:

 while (true) { var message = queue.Read(); Process(message); } 

What you could do to get a very simple fault tolerance is to retry after a failure:

 while (true) { for (i = 0; i < 3; i++) { try { var message = queue.Read(); Process(message); break; //exit for loop } catch (Exception ex) { //log //no throw = for loop runs the next attempt } } } 

Of course, you can just catch the db exceptions (or rather transaction failures) to just play those messages.

Microservices

I know Micro-service is a sound word. But in this case, this is a great solution. Instead of having a monolithic kernel that processes all messages, split the application into smaller parts. Or in your case, just turn off the processing of certain types of messages.

If you have five nodes on which your application is running, you can make sure that node A receives messages related to orders, node B receives messages related to delivery, etc.

Thus, you can still scale your application horizontally, you do not get conflicts, and this requires little effort (several message queues and reconfiguration of each node).

Answer 3

If you want to always update the database and always correspond to units already processed, then you have several updates on the same mutable object.

To do this, you need to serialize updates for the same object. Either you do this by breaking down your data from manufacturers, or by accumulating events for an object in one queue, or by locking an object in the workplace using distributed locking or database-level locking.

You can use the actor model (in java / scala world with akka), which creates a message queue for each object or group of objects that process them in series.

UPDATED You can try the akka port for .net here too . Here you can find a good tutorial with examples about using akka in scala . But for general principles, you need to look more for [actor model]. However, it has flaws.

In the end, this refers to the separation of your data and the possibility of creating a unique specialized worker (which can be reused and / or restarted in the event of a failure) for a specific object.

Answer 4

I assume that you have the means to securely access the product queue in all production services. Given that one simple way to avoid conflict is to use global product queues next to the main queue

 // Queue[X] is the queue for product X // QueueMain is the main queue DoWork(ProductType X) { if (Queue[X].empty()) { product = QueueMain().pop() if (product.type != X) { Queue[product.type].push(product) return; } }else { product = Queue[X].pop() } //process product... } 

Queue access must be atomic

Answer 5

1) Every high-performance data solution that I can come up with has something built-in to deal with just such a conflict. Details will depend on your final choice for data storage. In the case of a traditional relational database, this was baked without any further action on your part. For details, see Selected Technical Documentation.

2) Understand your data model and usage patterns. Build your data warehouse accordingly. Do not plan a scale that you will not have. Optimize your most common usage patterns.

3) Challenge your assumptions. Do you really have to mutate the same object very often from multiple roles? Sometimes the answer is yes, but often you can just create a new object that looks like a display. IE, use the / logging method instead of a single entity approach. Ultimately, large volumes of updates on a single object will never scale.