How to design a service that processes messages arriving in a queue

Question

I have a design question for a multi-threaded windows service that processes messages from multiple clients. The rules are

• Each message is to process something for an entity (with a unique id) and can be different i.e DoA, DoB, DoC etc. Entity id is in the payload of the message.
• The processing may take some time (up to few seconds).
• Messages must be processed in the order they arrive for each entity (with same id).
• Messages can however be processed for another entity concurrently (i.e as long as they are not the same entity id)
• The no of concurrent processing is configurable (generally 8)
• Messages can not be lost. If there is an error in processing a message then that message and all other messages for the same entity must be stored for future processing manually.
• The messages arrive in a transactional MSMQ queue.

How would you design the service. I have a working solution but would like to know how others would tackle this.

Answer 1

While my requirements were different from yours, I did have to deal with the concurrent processing from a message queue. My solution was to have a service which would look at each incoming message and hand it off to an agent process to consume. The service has a setting which controls how many agents it can have running.

Answer 2

I would look at having n thread each that read from a single thread-safe queue. I would then hash the EntityId to decide witch queue on put an incomming message on.

Sometimes, some threads will have nothing to do, but is this a problem if you have a few more threads then CPUs?

(Also you may wish to group entites by type into the queues so as to reduce the number of locking conflits in your database.)

Answer 3

My approach would be the following:

1. Create a threadpool with your configurable number of threads.
2. Keep map of entity ids and associate each id with a queue of messages.
3. When you receive a message place it in the queue of the corresponding entity id.
4. Each thread will only look at the entity id dedicated to it (e.g. make a class that is initialized as such Service(EntityID id)).
5. Let the thread only process messages from the queue of its dedicated entity id.
6. Once all the messages are processed for the given entity id remove the id from the map and exit the loop of the thread.
7. If there is room in the threadpool, then add a new thread to deal with the next available entity id.

You'll have to manage the messages that can't be processed at the time, including the situations where the message processing fails. Create a backlog of messages, etc.

If you have access to a concurrent map (a lock-free/wait-free map), then you can have multiple readers and writers to the map without the need of locking or waiting. If you can't get a concurrent map, then all the contingency will be on the map: whenever you add messages to a queue in the map or you add new entity id's you have to lock it. The best thing to do is wrap the map in a structure that offers methods for reading and writing with appropriate locking.

I don't think you will see any significant performance impact from locking, but if you do start seeing one I would suggest that you create your own lock-free hash map: http://www.azulsystems.com/events/javaone_2007/2007_LockFreeHash.pdf

Implementing this system will not be a rudimentary task, so take my comments as a general guideline... it's up to the engineer to implement the ideas that apply.

Answer 4

First thing you do is step back, and think about how critical is performance for this application. Do you really need to proccess messages concurrently? Is it mission critical? Or do you just think that you need it? Have you run a profiler on your service to find the real bottlenecks of the procces and optimized those?

The reason I ask, is be cause you mention you want 8 concurrent procceses - however, if you make this app single threaded, it will greatly reduce the complexity & developement & testing time... And since you only want 8, it almost seems not worth it...

Secondly, since you can only proccess concurrent messages on the same entity - how often will you really get concurrent requests from your client to procces the same entity? Is it worth adding so many layers of complexity for a use case that might not come up very often?

I would KISS. I'd use MSMQ via WCF, and keep my WCF service as a singleton. Now you have the power, ordered reliability of MSMQ and you are now meeting your actual requirements. Then I'd test it at high load with realistic data, and run a profiler to find bottlenecks if i found it was too slow. Only then would I go through all the extra trouble of building a much more complex app to manage concurrency for only specific use cases...

One design to consider is creating a central 'gate keeper' or 'service bus' service who receives all the messages from the clients, and then passes these messages down to the actual worker service(s). When he gets a request, he then finds if another one of his clients are already proccessing a message for the same entity - if so, he sends it to that same service he sent the other message to. This way you can proccess the same messages for a given entity concurrently and nothing more... And you have ease of seamless scalability... However, I would only do this if I absolutely had to and it was proved out via profiling and testing, and not because 'we think we needed it' (see YAGNI principal :))