MQ中传输大消息时出现的OOM问题

java.lang.OutOfMemoryError: Java heap space
Dumping heap to java_pid5328.hprof ...
Heap dump file created [529688434 bytes in 4.197 secs]
Exception in thread "EMQ Broker[broker-cnd] Scheduler" java.lang.OutOfMemoryError: Java heap space
        at com.xxx.xxx.store.kahadb.disk.journal.DataFileAccessor.readRecord(Unknown Source)
        at com.xxx.xxx.store.kahadb.disk.journal.Journal.read(Unknown Source)
        at com.xxx.xxx.store.kahadb.plist.PListStoreImpl.getPayload(Unknown Source)
        at com.xxx.xxx.store.kahadb.plist.PListImpl$PListIteratorImpl.next(Unknown Source)2014-06-06 14:56:31,169 | WARN  | Async error occurred:
java.lang.OutOfMemoryError: Java heap space
        at com.xxx.xxx.util.DataByteArrayOutputStream.ensureEnoughBuffer(Unknown Source)
        at com.xxx.xxx.util.DataByteArrayOutputStream.write(Unknown Source)
        at com.xxx.xxx.openwire.v6.BaseDataStreamMarshaller.looseMarshalByteSequence(Unknown Source)
        at com.xxx.xxx.openwire.v6.MessageMarshaller.looseMarshal(Unknown Source)
        at com.xxx.xxx.openwire.v6.EMQMessageMarshaller.looseMarshal(Unknown Source)
        at com.xxx.xxx.openwire.v6.EMQBytesMessageMarshaller.looseMarshal(Unknown Source)
        at com.xxx.xxx.openwire.OpenWireFormat.marshal(Unknown Source)
        at com.xxx.xxx.broker.region.cursors.FilePendingMessageCursor.getByteSequence(Unknown Source)
        at com.xxx.xxx.broker.region.cursors.FilePendingMessageCursor.tryAddMessageLast(Unknown Source)
        at com.xxx.xxx.broker.region.cursors.FilePendingMessageCursor.addMessageLast(Unknown Source)
        at com.xxx.xxx.broker.region.Queue.sendMessage(Unknown Source)
        at com.xxx.xxx.broker.region.Queue.doMessageSend(Unknown Source)
        at com.xxx.xxx.broker.region.Queue.send(Unknown Source)
        at com.xxx.xxx.broker.region.AbstractRegion.send(Unknown Source)
        at com.xxx.xxx.broker.region.RegionBroker.send(Unknown Source)
        at com.xxx.xxx.broker.jmx.ManagedRegionBroker.send(Unknown Source)
        at com.xxx.xxx.broker.BrokerFilter.send(Unknown Source)
        at com.xxx.xxx.broker.CompositeDestinationBroker.send(Unknown Source)
        at com.xxx.xxx.broker.TransactionBroker.send(Unknown Source)
        at com.xxx.xxx.broker.BrokerFilter.send(Unknown Source)
        at com.xxx.xxx.broker.MutableBrokerFilter.send(Unknown Source)
        at com.xxx.xxx.broker.TransportConnection.processMessage(Unknown Source)
        at com.xxx.xxx.command.EMQMessage.visit(Unknown Source)
        at com.xxx.xxx.broker.TransportConnection.service(Unknown Source)
        at com.xxx.xxx.broker.TransportConnection$1.onCommand(Unknown Source)
        at com.xxx.xxx.transport.ResponseCorrelator.onCommand(Unknown Source)
        at com.xxx.xxx.transport.MutexTransport.onCommand(Unknown Source)
        at com.xxx.xxx.transport.vm.VMTransport.iterate(Unknown Source)
        at com.xxx.xxx.thread.PooledTaskRunner.runTask(Unknown Source)
        at com.xxx.xxx.thread.PooledTaskRunner$1.run(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)

        at com.xxx.xxx.store.kahadb.plist.PListImpl$PListIteratorImpl.next(Unknown Source)
        at com.xxx.xxx.broker.region.cursors.FilePendingMessageCursor$DiskIterator.next(Unknown Source)
        at com.xxx.xxx.broker.region.cursors.FilePendingMessageCursor$DiskIterator.next(Unknown Source)
        at com.xxx.xxx.broker.region.cursors.FilePendingMessageCursor.next(Unknown Source)
        at com.xxx.xxx.broker.region.Queue.doPageInForDispatch(Unknown Source)
        at com.xxx.xxx.broker.region.Queue.pageInMessages(Unknown Source)
        at com.xxx.xxx.broker.region.Queue.doBrowse(Unknown Source)
        at com.xxx.xxx.broker.region.Queue.expireMessages(Unknown Source)
        at com.xxx.xxx.broker.region.Queue.access$100(Unknown Source)
        at com.xxx.xxx.broker.region.Queue$2.run(Unknown Source)
        at com.xxx.xxx.thread.SchedulerTimerTask.run(Unknown Source)
        at java.util.TimerThread.mainLoop(Timer.java:512)
        at java.util.TimerThread.run(Timer.java:462)
************************************
【分析】
1、检查是否为游标配置和代码中没有调用clearBody()的问题导致的OOM。
2、消息游标占用正常、从分析工具（ISA -->HeapAnalyzer）看是内存中放置的byteMessage的个数，导致老代被用完了。
3、考虑调整发送byteMessage的代码，产生的byteMessage用完后置空。
4、游标对内存的占用，可以合理控制；发现如果单节点切分byteMessage时，也会导致oom，此时并没有游标在使用，游标仅记录待转发消息的数量。
5、就单节点切分ByteMessage进行分析，发现是因为：com.xxx.xxx.broker.region.Queue.doBrowse触发的问题。因此考虑将队列中的配置策略参数，关于browse的调小。
   即：maxBrowsePageSize和maxPageSize配置为1。
6、实际触发是由过期消息检查com.xxx.xxx.broker.region.Queue.expireMessages引起的，可以考虑过期消息检查时，browse大类型消息时，限制browse大小。
   ===>是否可行？==>只能对某个队列限制，而队列中的消息类型不能进行限制。
注意：消息入队时才占用游标，当消息进入队列后，就不占用游标了。
7、CMS算法会导致碎片。==>
8、默认的【MAX_PAGE_SIZE】与【MAX_BROWSE_PAGE_SIZE】大小关系；
   public static final int MAX_PAGE_SIZE = 200;
   public static final int MAX_BROWSE_PAGE_SIZE = MAX_PAGE_SIZE * 2;
9、最终解决方法：在代码中判断是否为特定队列，则将其参数maxBrowsePageSize和maxPageSize设置为1，其实关键参数是后者，即【maxPageSize】。
==>maxBrowsePageSize参数用于通过jmx调用Queue的MBean方法，browse时，每次最多显示的消息数量；
==>maxPageSize参数用于从持久化导入消息到内存时，每次的页导入的消息数量。
10、在【com.xxx.xxx.broker.region.Queue.doPageInForDispatch】
/////////begin///////
int toPageIn = Math.min(getMaxPageSize(), messages.size());
/////////end/////////

【温馨提示】
如果您觉得满意，可以选择支持下，您的支持是我最大的动力：