Akka, Camel and ActiveMQ: throttling consumers

Question

I've got a very basic skeleton Scala application (with Akka, Camel and ActiveMQ) where I want to publish onto an ActiveMQ queue as quickly as possible, but then only consume from that queue at a particular rate (eg. 1 per second).

Here's some code to illustrate that:

MyProducer.scala

class Producer extends Actor with Producer with Oneway {
  def endpointUri = "activemq:myqueue"
}

MyConsumer.scala

class MyConsumer extends Actor with Consumer {

  def endpointUri = "activemq:myqueue"

  def receive = {
    case msg: CamelMessage => println("Ping!")
  }
}

In my main method, I then have all the boilerplate to set up Camel and get it talking to ActiveMQ, and then I have:

// Start the consumer
val consumer = system.actorOf(Props[MyConsumer])
val producer = system.actorOf(Props[MyProducer])

// Imagine I call this line 100+ times
producer ! "message"

How can I make it so that MyProducer sends things to ActiveMQ as quickly as possible (ie. no throttling) whilst making sure that MyConsumer only reads a message every x seconds? I'd like each message to stay on the ActiveMQ queue until the last possible moment (ie. when it's read by MyConsumer).

So far, I've managed to use a TimerBasedThrottler to consume at a certain rate, but this still consumes all of the messages in one big go.

Apologies if I've missed something along the way, I'm relatively new to Akka/Camel.

You shouldn't create a new producer every time. Just create one and store the ActorRef. — Ryan
Thanks Ryan, good spot - I'm now only instantiating it once. — Sam Starling

WestCoastProjects WestCoastProjects · Accepted Answer · 2014-05-15T03:16:30

How many consumers comprise "MyConsumer"?

a) If it were only one, then it is unclear why a simple sleep between reading/consuming messages would not work.

If there are multiple consumers, which behavior are you requiring:

each consumer is throttled to the specified consumption rate. In that case each Consumer thread still behaves as mentioned in a)
the overall pool of consumers is throttled to the consumption rate. In that case a central Throttler would need to retain the inter-message delay and block each consumer until the required delay were met . There would be the complexity of managing when there were backlogs - to allow "catch-up". You probably get the drift here.

It may be you were looking for something else /more specific in this question. If so then please elaborate.

Akka, Camel and ActiveMQ: throttling consumers

1 Answers