Throttled processing of ActiveMQ with C#

Question

We have a broker which receives a lot of messages and we want to use a Windows Service to consume those messages.

The Windows Service will contain one Task for receiving those messages and around 50-70 tasks for handling the messages. The number of consuming tasks must be limited, because we have limitations of 3rd party elements. But we will run multiple Windows Service process for scaling.

The ActiveMQ .NET provider just offers an event for new messages.

What would be the best way in this scenario for an hand-over of the messages from Producer to Consumer? If there is a new message receiving from ActiveMQ, the Producer should wait until a Consumer is ready for handling a new message.

Should we write a kind of LimitedQueue or is there anything built-in in TPL?

Answer 1

I think you could use here a BlockingCollection with boundedCapacity 1:

BlockingQueue<YourMessage> queue = new BlockingCollection<YourMessage>(1);

Now your Tasks can consume messages with

foreach(var message in queue.GetConsumingEnumerable())

If no message in queue, all tasks will wait non-busy.

Add will block until queue has space for new entries. Use TryAdd if you want to use a timeout for your message adding.