scala – 使用Akka Actors的OutOfMemoryError
发布时间:2020-12-16 18:36:31 所属栏目:安全 来源:网络整理
导读:我有一个使用RabbitMQ消息的应用程序,我正在使用Actors来处理工作. 这是我的方法: object QueueConsumer extends Queue { def consumeMessages = { setupListener(buildChannel(resultsQueueName),resultsQueueName,resultsCallback) } private def setupLi
我有一个使用RabbitMQ消息的应用程序,我正在使用Actors来处理工作.
这是我的方法: 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) } } } 运行几秒钟后,它会抛出java.lang.OutOfMemoryError:超出GC开销限制. 我认为它正在发生,因为我正在为我收到的每条消息开始一个新的Actor – 所以如果我有100000条消息,它将创建10万个演员.这是一个好方法还是我应该实现像’演员池’这样的东西? 任何人都知道如何在我的场景中避免OutOfMemoryError? 预先感谢. EDIT1: 改变方法: 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,consumer) while (true) { val delivery = consumer.nextDelivery() val msg = new String(delivery.getBody()) context.actorOf(Queue2.props(msg)) channel.basicAck(delivery.getEnvelope.getDeliveryTag,false) } } } 解决方法
你的每个消息的演员在完成后需要自己停止,否则他们会永远呆在那里.请参阅
Actor lifecycle和
stopping Actors上的文档.在这里,您只需要在处理完成后添加context.stop(self).
(编辑:李大同) 【声明】本站内容均来自网络,其相关言论仅代表作者个人观点,不代表本站立场。若无意侵犯到您的权利,请及时与联系站长删除相关内容! |