CODEWITHSUNDEEP
c#.netmessage-queue
user1713059
user1713059

Reputation: 1529

Synchronized message queue

I have a system where a set of operations need to be done, each operation has a type and an entity. Operations for each entity can be done independently, but within entity the order of types does matter. Operations on a type can start only when all operations of the previous type have finished. Operations usually take some data from database, process it, store results in the database again.

How can I achieve this effect (preferably within the .NET world) assuming the amount of operations will require use of many machines in parallel?

P.S. Right now I am using service bus, each message is an operation-entity-type combination. This does not meet my needs at all since the order is not enforced and it's not clear when an operation connected with a message has actually finished.

EDIT: I need to find a logical way of processing those operations in the right order keeping everything (which can be) parallel.

Example: I need to do 3 types of operations (T1-T3) for 2 entities (E1, E2). There can be many operations having the same type and entity.

-- Entity 1
[T1, E1][T1, E1][T1, E1][T1, E1] - 4 operations Type 1, Entity 1
[T2, E1][T2, E1] - 2 ops, Type 2, Entity 1
[T3, E1] - 1 operation, Type 3, Entity 1
-- Entity 2
[T2, E2][T2, E2]
[T3, E2]

Generally I can process entity 1 in parallel to entity 2. I can do each specific [Tx, Ex] in parallel too (for example all [T2, E1]). I need to keep order of types, T1->T2->T3. T2 operations need to wait until all T1 operations have finished within an entity.

Upvotes: 1

Views: 195

Answers (2)

Glubus
Glubus

Reputation: 2855

How about you keep a state in your entity, that induces which types of operations are allowed to be performed.

You'll have some worker that will distribute the operations over different threads, and will change the state when there are no more operations of a certain type to be done.

I think it's possible to check if started threads are finished, but you might even want to add state to the operations too, (e.g. Queued, InProgress, Done).

You can just query if an entity still has operations of a certain type that are not 'Done', and when they are all Done, update the state of the Entity, allowing the next type of operation with state Queued to be dequeued and processed.

Note

This only works if the invariant holds that it is not possible to go from one state to a previous one, meaning that this only works when it is not possible for an entity to process T1 operations after it has done T2 operations. In practice this means that it should not be allowed that an entity gets new operations assigned after it has started being processed.

Upvotes: 1

MichalO
MichalO

Reputation: 37

Maybe you should use separate queues for each type of operation (message). You process this queues in order and only if previous queue is empty. Eg. there are 3 queues A, B and C. Consumer checks if there is any message in A queue. If it is - consumer gets it and process. If not - it's checks B queue. And so on. But there still could be a problem with last message from A queue - it may still process when messages from B queue will start.

Similar solutions - use message priority. But problem with last message still remains.

Upvotes: 1

Related Questions