I wanted to write about real use case of less known design patterns and how they can help achieving a simplified design and high performance.
In message oriented business use cases, sometimes there is a need to process messages in the received order so that system is consistent according to the time.
For example if trader buys 10 lot of a share at 11:00:00 am and then wants to buy another 15 lot of the share at 11:00:01. Order management system has to process these orders in the received order to make sure trader’s account is updated correctly and system does check remaining funds, purchased shares in a right way.
Another similar use case is processing events of a football match. In a match there is a number of events from scoring a goal, yellow cards, red cards, stop times, free kicks, etc. In a betting portal all these events could be used as a betting market, a market operator may create a betting market for number of free kicks or when/how many yellow/red cards will happen.
A simple use case scenario gets complicated when number of trade orders/ betting market events are in high volume and there is business demand to settle these activities instantly.
A naive approach to process orders from a message queue is to create a multi-threaded executor which receives messages and dispatches the them to the underlying threads. By default a thread executor doesn’t have any logic of dispatching work items to the worker threads.
Let’s consider a trader places two trade orders M1, M4 and expects that system process these trades in the received order while another trader places M2 and M5 orders.
How to ensure trades will be processed in the received order in a multi-threaded environment ?
In the naive approach unfortunately there is not guarantee that thread2 will be executed after thread1 which processes M1.
Following test case and console output shows unordered message processing.
Unordered message processing
Not so good solution:
To solve this problem, there is a need to mark relevant messages together. For example adding a trader id field to the message as a correlation id helps identifying which messages are for a particular trader.
One approach to solve this problem is creating a map of correlation id and thread id, so that a customized thread executor can figure out how to distribute messages amongst worker threads.
Following code piece demonstrates that approach.
Map implementation can be a self-expiring map, like a cache, to ensure map doesn’t grow over the time.
However this solution introduces a complicated expiring map and how to determine expiration of map items.
A better solution:
The following picture illustrates a better solution. This solution distributes messages having same correlation id to the same underlying thread. Once messages are assigned to the same thread, thread processes these messages in the received order.
It is easier to use hash value of the correlation id and size of the thread executors to determine which executor should pick the message. In this solution it is assumed correlation is immutable once being assigned.
As hash value of the correlation id is fixed, same thread/executor can be assigned to the messages having same correlation id.
Following code piece shows how hash value is used to find index of a thread/executor.
Whole project can be found on github