ActiveMQ вышел из памяти при отправке больших файлов по очереди

Question

При передаче сообщений размера 100MB с использованием очереди, ActiveMQ впадающим отказа от ошибки памяти, мы используем файл курсор для очереди -

Очереди Detail - Наш продюсера отправляет Непрерывные сообщения размером 100 МБ на сообщения, а продюсер будет продолжать производить через цикл while из тех же 100 МБ сообщений.

Мы используем размер кучи по умолчанию, который поставляется с activeMQ с макс. 1 ГБ.

У нас есть настройки конфигурации ActiveMQ следующим образом:

<policyEntry queue=">" producerFlowControl="false" memoryLimit="512mb" maxPageSize="1000000">    
<pendingQueuePolicy>        
<fileQueueCursor /> 
</pendingQueuePolicy> 
</policyEntry> 

На потребляем стороне мы имеем потребителя Async, который будет держать прослушивание сообщений, поступающих и отправить авто-Ack.

После этой программы работает некоторое время ActiveMQ выдает следующее сообщение об ошибке:

2016-04-21 14:52:18,961 | ERROR | Error in thread 'ActiveMQ BrokerService.worker.1' | org.apache.activemq.broker.BrokerService | ActiveMQ BrokerService.worker.1 
java.lang.OutOfMemoryError: Java heap space 
    at org.apache.activemq.util.DataByteArrayOutputStream.ensureEnoughBuffer(DataByteArrayOutputStream.java:249)[activemq-client-5.13.1.jar:5.13.1] 
    at org.apache.activemq.util.DataByteArrayOutputStream.writeBoolean(DataByteArrayOutputStream.java:140)[activemq-client-5.13.1.jar:5.13.1] 
    at org.apache.activemq.openwire.v11.BaseDataStreamMarshaller.looseMarshalByteSequence(BaseDataStreamMarshaller.java:627)[activemq-client-5.13.1.jar:5.13.1] 
    at org.apache.activemq.openwire.v11.MessageMarshaller.looseMarshal(MessageMarshaller.java:300)[activemq-client-5.13.1.jar:5.13.1] 
    at org.apache.activemq.openwire.v11.ActiveMQMessageMarshaller.looseMarshal(ActiveMQMessageMarshaller.java:111)[activemq-client-5.13.1.jar:5.13.1] 
    at org.apache.activemq.openwire.v11.ActiveMQTextMessageMarshaller.looseMarshal(ActiveMQTextMessageMarshaller.java:111)[activemq-client-5.13.1.jar:5.13.1] 
    at org.apache.activemq.openwire.OpenWireFormat.marshal(OpenWireFormat.java:161)[activemq-client-5.13.1.jar:5.13.1] 
    at org.apache.activemq.broker.region.cursors.FilePendingMessageCursor.getByteSequence(FilePendingMessageCursor.java:480)[activemq-broker-5.13.1.jar:5.13.1] 
    at org.apache.activemq.broker.region.cursors.FilePendingMessageCursor.flushToDisk(FilePendingMessageCursor.java:440)[activemq-broker-5.13.1.jar:5.13.1] 
    at org.apache.activemq.broker.region.cursors.FilePendingMessageCursor.onUsageChanged(FilePendingMessageCursor.java:401)[activemq-broker-5.13.1.jar:5.13.1] 
    at org.apache.activemq.usage.Usage$1.run(Usage.java:308)[activemq-client-5.13.1.jar:5.13.1] 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)[:1.8.0_74] 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)[:1.8.0_74] 
    at java.lang.Thread.run(Thread.java:745)[:1.8.0_74] 

Кто-нибудь знает, как решить эту проблему? Кажется, что это не происходит, когда я отправляю сообщения меньшего размера, например, сообщения размером менее 10 МБ.

Answer 1

Непрерывные сообщения хранятся в памяти, а не сохраняются в хранилище данных, как описано here. Таким образом, 1 ГБ скоро исчезнет, ​​особенно если вы не можете потреблять так быстро, как вы производите.

Конечно, вы можете увеличить объем памяти, выделенной ActiveMQ в файле activemq.xml, но вам может быть лучше не работать, даже если вам не нужно восстанавливаться, и, возможно, срок действия сообщений истекает через некоторое время эмулировать ненасыщенность (при необходимости).

Я бы предложил другие решения, такие как разбиение сообщений на что-то более управляемое или использование общего хранилища файлов для данных и отправка сообщений, содержащих указатели на данные. Помимо издержек на обработку для ActiveMQ, полагайте, что у вас больше, чем обычно, влияние сети (например, если у вас есть безопасная связь с вашим экземпляром AMQ, вы шифруете/дешифруете сообщения 100M, а не дешевые).