Concurrent Processing of Messages from a RabbitMq Queue

[Akinnagbe]

I have a .net core background service that processes messages from a RabbitMq Queue, the messages can grow to thousands depending on the peak period, hence I have to manually instantiate each background service to process the messages one after the other.

My question is how can I use Actor Model to process the messages concurrently as each message arrives in the queue.

Please help.

[Danthar]

The simplest thing that can work, if you care about consumer ack (RMQ). is by retrieving an IActorRef from somewhere. And then doing an Ask<> on that actorref. Your actor processed the message, and sends a response through the Ask to indicate its done. Your consumer stops blocking/waiting on the ask, completes and your consumer ack is send back to RMQ and the message is removed from the queue.

If you dont care about consumer acks (RMQ). Do the same things except use Tell instead of ask. Then your consumer will complete right away possibly even before the message is actually processed.

With regards to concurrent processing. It depends on what library you are using to connect with RMQ. But you can configure most RMQ clients as to how many messages it should process at a time for a specific queue/listener (also known as a channel in RMQ client spreak).

Masstransit has these options. the default RabbitMq.Client package has these options. And other client libs that i know have them.

There is also the Alpakka amqp streaming integration packages. That you can use to directly interface with RMQ with a Akka.Stream Source.

https://github.com/akkadotnet/Alpakka/tree/dev/src/Amqp