OutOfMemoryError using Akka Actors

Question

I have an application that consumes messages from RabbitMQ, and I use Actors to handle the work.

Here is my approach:

object QueueConsumer extends Queue { def consumeMessages = { setupListener(buildChannel(resultsQueueName), resultsQueueName, resultsCallback) } private def setupListener(receivingChannel: Channel, queue: String, f: (String) => Any) { Akka.system.scheduler.scheduleOnce(Duration(10, TimeUnit.SECONDS), Akka.system.actorOf(Props(new QueueActor(receivingChannel, queue, f))), "") } } class QueueActor(channel:Channel, queue:String, f:(String) => Any) extends Actor{ def receive = { case _ => startReceiving } def startReceiving = { val consumer = new QueueingConsumer(channel) channel.basicConsume(queue, false, consumer) while (true) { val delivery = consumer.nextDelivery() val msg = new String(delivery.getBody()) context.actorOf(Props(new Actor { def receive = { case some: String => f(some) } })) ! msg channel.basicAck(delivery.getEnvelope.getDeliveryTag, false) } } } 

After a few seconds, it throws the java.lang.OutOfMemoryError: GC upper limit exceeded . .

I think this is because I am starting a new actor for every message I receive - so if I have 100,000 messages, this will create 100,000 actors. Is this a good approach or should I implement something like an "actor pool"?

Does anyone have an idea how I can avoid OutOfMemoryError in my script?

Thanks in advance.

edit1:

changed approach to:

 class Queue2(json:String) extends Actor { def receive = { case x: String => val envelope = MessageEnvelopeParser.toObject(x) val processor = ProcessQueueServiceFactory.getProcessResultsService() envelope.messages.foreach(message => processor.process(message)) } } object Queue2 { def props(json: String): Props = Props(new Queue2(json)) } class QueueActor(channel:Channel, queue:String) extends Actor { def receive = { case _ => startReceiving } def startReceiving = { val consumer = new QueueingConsumer(channel) channel.basicConsume(queue, false, consumer) while (true) { val delivery = consumer.nextDelivery() val msg = new String(delivery.getBody()) context.actorOf(Queue2.props(msg)) channel.basicAck(delivery.getEnvelope.getDeliveryTag, false) } } } 

Answer 1

Your characters from one message will have to stop themselves when done, otherwise they will remain forever. Please review the actor’s life cycle documents and stop the participants . Here you just need to add context.stop(self) after processing is complete.