pzxsheng

          有種相見不敢見的傷痛,有種愛還埋藏在心中

          activeMQ指南針_Queue完整分析

          原文地址: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的使用,在內存使用方面非常慎重,任何消息只有在內存里有空閑區域時,才能放到內存里,之后才能發給消費者。當消息被消費者消耗掉了后,確認信息會發給activeMQQueue接收到這些確認消息后,會把那些被確認的消息所占用的內存釋放掉。

           

          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、功能改進,都可以在這個計劃里交流。所有在笑臉計劃中提出的問題、功能改進、解決方案,都將完全通過網絡無償分享給所有人。

          posted on 2013-03-15 16:24 科菱財神 閱讀(715) 評論(0)  編輯  收藏 所屬分類: ActiveMQ

          導航

          <2013年3月>
          242526272812
          3456789
          10111213141516
          17181920212223
          24252627282930
          31123456

          統計

          常用鏈接

          留言簿(1)

          隨筆分類

          隨筆檔案

          搜索

          最新評論

          閱讀排行榜

          評論排行榜

          主站蜘蛛池模板: 漠河县| 大同市| 虎林市| 西林县| 灵山县| 襄垣县| 邛崃市| 鄯善县| 额尔古纳市| 巴楚县| 永定县| 金堂县| 邻水| 平陆县| 乐清市| 大邑县| 长宁县| 陆良县| 攀枝花市| 乐至县| 东方市| 南昌市| 南汇区| 台湾省| 泸定县| 宝应县| 石景山区| 清新县| 福建省| 千阳县| 鱼台县| 舒城县| 黎平县| 汨罗市| 庆城县| 健康| 抚州市| 饶平县| 蒙山县| 凯里市| 台中市|