原文地址:http://www.360doc.com/content/13/0315/16/11765546_271697514.shtml
在接觸activeMQ的這一段時間里,我們還是保持開始對它的態度,它是個優秀的開源消息中間件。消息中間件是個非常重要的搭建企業應用系統的重要組件,我們在不斷深入分析activeMQ的過程中,發現直到5.1這個版本,都還是存在不少問題,有些是很致命,但正因為如此,我們更加堅定了要全面掌握activeMQ,我們不想重新做“輪子”,但我們要具備在輪子壞了或不好用的情況下,要能獨立解決碰到的這些問題。下面我們通過分析網友提出的一個典型的問題場景,來作為我們指南針計劃的結束。
Queue作為activeMQ里面一個很重要的通訊方式,網友的場景如下:
測試queue持久化消息時,發送接收20W條消息。打開消息消費者,連上再斷開,反復進行這步操作,能接收到消息,接收端有時候會阻塞,但不能完全接收完20W條消息。(其實5000條就會發生問題,不用20W這么多)
相關背景知識:
因為這是5.1版本的一個非常嚴重的bug,所以我們會比較詳細的進行分析。(我們在最終解決問題后,上activeMQ官網上發現它最新的源碼是解決了該問題的,但這并不影響這個問題的典型性)。下面我們將從3個方面來分析:Queue消息的接收和發送、內存使用機制、消息的審查(audit)、消息在文件中的存儲機制。
l Queue消息的接收和發送

Queue接收消息并發給需要的消費者,具體過程如下:
1. Queue從消息生產者接收消息。
2. Queue使用一個“存儲指針”來接收這些消息。當內存有空閑區域時,“存儲指針”把消息放到內存中,當內存不夠時,則把消息們存入磁盤文件。
3. 當有活動的(active)的消息消費者時,Queue會首先把“存儲指針”的內存中的消息送給消費者,當內存的消息被消費掉,則從磁盤文件中再讀入其他的消息(出問題處),直至消息都被消費掉了。
其中最關鍵的方法是Queue類里的doPageIn()。
l 內存使用機制
activeMQ為了適應企業級的365*24的使用,在內存使用方面非常慎重,任何消息只有在內存里有空閑區域時,才能放到內存里,之后才能發給消費者。當消息被消費者消耗掉了后,確認信息會發給activeMQ。Queue接收到這些確認消息后,會把那些被確認的消息所占用的內存釋放掉。
l 消息的審查(audit)
為了防止消息的重復發送,activeMQ采用了一個審查機制,它負責審查某條消息是否重復。它是一個最近最久未使用算法(LRU)隊列。每個隊列元素它是一個bit數組,它的運行機制如下所示:

消息是一個個按照順序進入bit數組,具體算法answer = (index - firstIndex) / BitArray.LONG_SIZE,其中:
BitArray.LONG_SIZE是每個bit數組的大小。
Index是消息的編號。(它是按照+1順序增加的)
firstIndex是整個LRU隊列的首Index,這個值會經常變化,因為當達到LRU的上限時,老的一批就被清除了,firstIndex += BitArray.LONG_SIZE。(出問題處)
l 消息在文件中的存儲機制
存放在文件中的消息,它們是按照如下方式進行組織的:

每個消息都知道它的上一個和下一個消息,當它自身被刪除后,相應的關系會進行調整。
問題原因分析:
因為activeMQ在編碼實現的時候,原本的想法應該是這樣的:
1. 從生產者接收消息,如果Queue有可用的內存就放在內存中,沒有則存入文件中。
2. Queue發送消息給消費者時,先發送已經保存在內存中的消息。
3. 當內存中消息發送完后,順序讀入(這里是關鍵)文件中的消息,通過消息的審查機制,確認不是重復消息,則放入內存中供后續操作使用。
但是activeMQ5.1版本的實現,問題就出在第三步的順序讀入。因為從文件中讀入它有個先決條件,那就是必須要有可用的內存,如果沒有可用的話,就放棄本次消息讀入,并且應該放棄這次讀取操作。但是5.1版本是繼續往下讀,這就導致順序錯亂,使得當內存可用的時候,讀入的消息在進行審查的時候,發生錯誤,錯誤認為它們是重復消息。這就導致發送20W條消息,不能保證完全收到。
解決方案:
類KahaReferenceStore的方法recoverNextMessages里的
if (entry != null) {
int count = 0;
do {
ReferenceRecord msg = messageContainer.getValue(entry);
if (msg != null ) {
if ( recoverReference(listener, msg)) {
count++;
lastBatchId = msg.getMessageId();
}
} else {
lastBatchId = null;
}
batchEntry = entry;
entry = messageContainer.getNext(entry);
} while (entry != null && count < maxReturned && listener.hasSpace());
}
改為
if (entry != null) {
int count = 0;
do {
ReferenceRecord msg = messageContainer.getValue(entry);
testTheNextMsgId(msg.getMessageId().toString());
if (msg != null )
{
if ( recoverReference(listener, msg))
{
count++;
lastBatchId = msg.getMessageId();
batchEntry = entry;
entry = messageContainer.getNext(entry);
}
else
{
break;
}
}
else
{
lastBatchId = null;
batchEntry = entry;
entry = messageContainer.getNext(entry);
}
} while (entry != null && count < maxReturned && listener.hasSpace());
}
activeMQ指南針計劃的結束,但它又是個新開始,我們通過這個計劃收獲了我們想要的東西了,同時我們不僅為各位朋友答疑解疑,也提供了activemqSpanner這個工具作為消息網絡拓撲圖工具。再一次感謝各位朋友對我們的信任。
現在,我們正式啟動activeMQ笑臉計劃。它的目的不再是給大家提供解決問題的方向,而是直接解決大家碰到的各種問題,給大家帶去笑臉。它將是一個長期堅持的事情,任何關于activeMQ使用過程的疑惑、問題、bug、功能改進,都可以在這個計劃里交流。所有在笑臉計劃中提出的問題、功能改進、解決方案,都將完全通過網絡無償分享給所有人。