一個topic就是一個主題。一個系統中，我們可以對消息劃分為一些topic，這樣我們就能通過topic，将消息發送到不同的queue。
一個topic下，我們可以設定多個queue，每個queue就是我們平時所說的消息隊列；因為queue是完全從屬于某個特定的topic的，是以當我們要發送消息時，總是要指定該消息所屬的topic是什麼。然後equeue就能知道該topic下有幾個queue了。但是到底發送到哪個queue呢？比如一個topic下有4個queue，那對于這個topic下的消息，發送時，到底該發送到哪個queue呢？那必定有個消息被路由的過程。目前equeue的做法是在發送一個消息時，需要使用者指定這個消息對應的topic以及一個用來路由的一個object類型的參數。equeue會根據topic得到所有的queue，然後根據該object參數通過hash code然後取模queue的個數最後得到要發送的queue的編号，進而知道該發送到哪個queue。這個路由消息的過程是在發送消息的這一方做的，也就是下面要說的producer。之是以不在消息伺服器上做是因為這樣可以讓使用者自己決定該如何路由消息，具有更大的靈活性。
就是消息隊列的生産者。我們知道，消息隊列的本質就是實作了publish-subscribe的模式，即生産者-消費者模式。生産者生産消息，消費者消費消息。是以這裡的producer就是用來生産和發送消息的。
就是消息隊列的消費者，一個消息可以有多個消費者。
消費者分組，這可能對大家來說是一個新概念。之是以要搞出一個消費者分組，是為了實作下面要說的叢集消費。一個消費者分組中包含了一些消費者，如果這些消費者是要叢集消費，那這些消費者會平均消費該分組中的消息。
equeue中的broker負責消息的中轉，即接收producer發送過來的消息，然後持久化消息到磁盤，然後接收consumer發送過來的拉取消息的請求，然後根據請求拉取相應的消息給consumer。是以，broker可以了解為消息隊列伺服器，提供消息的接收、存儲、拉取服務。可見，broker對于equeue來說是核心，它絕對不能挂，一旦挂了，那producer，consumer就無法實作publish-subscribe了。
叢集消費是指，一個consumer group下的consumer，平均消費topic下的queue。具體如何平均可以看一下下面的架構圖，這裡先用文字簡單描述一下。假如一個topic下有4個queue，然後目前有一個consumer group，該分組下有4個consumer，那每個consumer就被配置設定到該topic下的一個queue，這樣就達到了平均消費topic下的queue的目的。如果consumer group下隻有兩個consumer，那每個consumer就消費2個queue。如果有3個consumer，則第一個消費2個queue，後面兩個每個消費一個queue，進而達到盡量平均消費。是以，可以看出，我們應該盡量讓consumer group下的consumer的數目和topic的queue的數目一緻或成倍數關系。這樣每個consumer消費的queue的數量總是一樣的，這樣每個consumer伺服器的壓力才會差不多。目前前提是這個topic下的每個queue裡的消息的數量總是差不多多的。這點我們可以對消息根據某個使用者自己定義的key來進行hash路由來保證。
廣播消費是指一個consumer隻要訂閱了某個topic的消息，那它就會收到該topic下的所有queue裡的消息，而不管這個consumer的group是什麼。是以對于廣播消費來說，consumer group沒什麼實際意義。consumer可以在執行個體化時，我們可以指定是叢集消費還是廣播消費。
消費進度是指，當一個consumer group裡的consumer在消費某個queue裡的消息時，equeue是通過記錄消費位置(offset)來知道目前消費到哪裡了。以便該consumer重新開機後繼續從該位置開始消費。比如一個topic有4個queue，一個consumer group有4個consumer，則每個consumer配置設定到一個queue，然後每個consumer分别消費自己的queue裡的消息。equeue會分别記錄每個consumer對其queue的消費進度，進而保證每個consumer重新開機後知道下次從哪裡開始繼續消費。實際上，也許下次重新開機後不是由該consumer消費該queue了，而是由group裡的其他consumer消費了，這樣也沒關系，因為我們已經記錄了這個queue的消費位置了。是以可以看出，消費位置和consumer其實無關，消費位置完全是queue的一個屬性，用來記錄目前被消費到哪裡了。另外一點很重要的是，一個topic可以被多個consumer group裡的consumer訂閱。不同consumer group裡的consumer即便是消費同一個topic下的同一個queue，那消費進度也是分開存儲的。也就是說，不同的consumer group内的consumer的消費完全隔離，彼此不受影響。還有一點就是，對于叢集消費和廣播消費，消費進度持久化的地方是不同的，叢集消費的消費進度是放在broker，也就是消息隊列伺服器上的，而廣播消費的消費進度是存儲在consumer本地磁盤上的。之是以這樣設計是因為，對于叢集消費，由于一個queue的消費者可能會更換，因為consumer group下的consumer數量可能會增加或減少，然後就會重新計算每個consumer該消費的queue是哪些，這個能了解的把？是以，當出現一個queue的consumer變動的時候，新的consumer如何知道該從哪裡開始消費這個queue呢？如果這個queue的消費進度是存儲在前一個consumer伺服器上的，那就很難拿到這個消費進度了，因為有可能那個伺服器已經挂了，或者下架了，都有可能。而因為broker對于所有的consumer總是在服務的，是以，在叢集消費的情況下，被訂閱的topic的queue的消費位置是存儲在broker上的，存儲的時候按照不同的consumer group做隔離，以確定不同的consumer group下的consumer的消費進度互補影響。然後，對于廣播消費，由于不會出現一個queue的consumer會變動的情況，是以我們沒必要讓broker來儲存消費位置，是以是儲存在consumer自己的伺服器上。
通過上圖，我們能直覺的了解equeue。這個圖是從rocketmq的設計文檔中拿來的，呵呵。由于equeue的設計思想完全和rocketmq一緻，是以我就拿過來用了。每個producer可以向某個topic發消息，發送的時候根據某種路由政策（producer可自定義）将消息發送到某個特定的queue。然後consumer可以消費特定topic下的queue裡的消息。上圖中，topic_a有兩個消費者，這兩個消費者是在一個group裡，是以應該平均消費topic_a下的queue但由于有三個queue，是以第一個consumer分到了2個queue，第二個consumer分到了1個。對于topic_b，由于隻有一個消費者，那topic_b下的所有queue都由它消費。所有的topic資訊、queue資訊、還有消息本身，都存儲在broker伺服器上。這點上圖中沒有展現出來。上圖主要關注producer,consumer,topic,queue這四個東西之間的關系，并不關注實體伺服器的部署架構。
消息持久化方面主要考慮的是性能問題，還有就是消息如何快速的讀取。
1. 首先，一台broker上的消息不需要一直儲存在該broker伺服器上，因為這些消息總會被消費掉。根據阿裡rocketmq的設計，預設會1天删除一次已經被消費過的消息。是以，我們可以了解，broker上的消息應該不會無限制增長，因為會被定期删除。是以不必考慮一台broker上消息放不下的問題。
2. 如何快速的持久化消息？一般來說，我覺得有兩種方式：1）順序寫磁盤檔案；2）用現成的key,value的nosql産品來存儲；rocketmq目前用的是自己寫檔案的方式，這種方式的難點是寫檔案比較複雜，因為所有消息都是順序append到檔案末尾，雖然性能非常高，但複雜度也很高；比如所有消息不能全寫在一個檔案裡，一個檔案到達一定大小後需要拆分，一旦拆分就會産生很多問題，呵呵。拆分後如何讀取也是比較複雜的問題。還有由于是順序寫入檔案的，那我們還需要把每一個消息在檔案中的起始位置和長度需要記錄下來，這樣consumer在消費消息時，才能根據offset從檔案中拿到該消息。總之需要考慮的問題很多。如果是用nosql來持久化消息，那可以省去我們寫檔案時遇到的各種問題，我們隻需要關心如何把消息的key和該消息在queue中的offset對應起來即可。另外一點疑問是，queue裡的資訊要持久化嗎？先要想清楚queue裡放的是什麼東西。當broker接收到一個消息後，首先肯定是要先持久化，完成後需要把消息放入queue裡。但由于記憶體很有限，我們不可能把這個消息直接放入queue裡，我們其實要放的隻需要時該消息在nosql裡的key即可，或者如果是用檔案來持久化，那放的是該消息在檔案中的偏移量offset，即存儲在檔案的那個位置（比如是哪個行号）。是以，實際上，queue隻是一個消息的索引。那有必要持久化queue嗎？可以持久化，這樣畢竟在broker重新開機的時候，恢複queue的時間可以縮短。那需要和持久化消息同步持久化嗎？顯然不需要，我們可以異步定時持久化每個queue，然後恢複queue的時候，可以先從持久化的部分恢複，然後再把剩下的部分通過持久化的消息來補充以達到queue因為異步持久化而慢的部分可以追平。是以，經過上面的分析，消息本身都是放在nosql中，queue全部在記憶體中。
那消息如何持久化呢？我覺得最好的辦法是讓每個消息有一個全局的順序号，一旦消息被寫入nosql後，該消息的全局順序号就确定了，然後我們在更新對應的queue的資訊時，把該消息的全局順序号傳給queue，這樣queue就能把queue自己對該消息的本地順序号和該消息的全局順序号建立映射關系。相關代碼如下：
沒什麼比代碼更能說明問題了，呵呵。上的代碼的思路是，接收一個消息對象和一個queueid，queueid表示目前消息要放到第幾個queue裡。然後内部邏輯是，先擷取該消息的topic的所有queue，由于queue和topic都在記憶體，是以這裡沒性能問題。然後檢查一下目前傳遞進來的queueid是否合法。如果合法，那就定位到該queue，然後通過incrementcurrentoffset方法，将queue的内部序号加1并傳回，然後持久化消息，持久化的時候把queueid以及queueoffset一起持久化，完成後傳回一個消息的全局序列号。由于messagestore内部會把消息内容、queueid、queueoffset，以及消息的全局順序号一起作為一個整體儲存到nosql中，key就是消息的全局序列号，value就是前面說的整體（被序列化為二進制）。然後，在調用queue的setmessageoffset方法，把queueoffset和message的全局offset建立映射關系即可。最後傳回一個結果。messagestore.storemessage的記憶體實作大緻如下：
getnextoffset就是擷取下一個全局的消息序列号，queuemessage就是上面所說的“整體”，因為是記憶體實作，是以就用了一個concurrentdictionary來儲存一下queuemessage對象。如果是用nosql來實作messagestore，則這裡需要寫入nosql，key就是消息的全局序列号，value就是queuemessage的二進制序列化資料。通過上面的分析我們可以知道我們會将消息的全局序列号+queueid+queueoffset一起整體作為一條記錄持久化起來。這樣做有兩個非常好的特性：1）實作了消息持久化和消息在queue中的位置的持久化的原子事務；2）我們總是可以根據這些持久化的queuemessage還原出所有的queue的資訊，因為queuemessage裡包含了消息和消息在queue的中的位置資訊；
基于這樣的消息存儲，當某個consumer要消費某個位置的消息時，我們可以通過先通過queueid找到queue，然後通過消息在queueoffset(由consumer傳遞過來的)擷取消息的全局offset，然後根據該全局的offset作為key從nosql拿到消息。實際上現在的equeue是批量拉取消息的，也就是一次socket請求不是拉一個消息，而是拉一批，預設是32個消息。這樣consumer可以用更少的網絡請求拿到更多的消息，可以加快消息消費的速度。
producer在發送消息時，如何知道目前topic下有多少個queue呢？每次發送消息時都要去broker上查一下嗎？顯然不行，這樣發送消息的性能就上不去了。那怎麼辦呢？就是異步，呵呵。producer可以定時向broker發送請求，擷取topic下的queue數量，然後儲存起來。這樣每次producer在發送消息時，就隻要從本地緩存裡拿即可。因為broker上topic的queue的數量一般不會變化，是以這樣的緩存很有意義。那還有一個問題，目前producer第一次對某個topic發送消息時，queue哪裡來呢？因為定時線程不知道要向broker拿哪個topic下的queue數量，因為此時producer端還沒有一個topic呢，因為一個消息都還沒發送過。那就是需要判斷一下，如果目前topic沒有queue的count資訊，則直接從broker上擷取queue的count資訊。然後再緩存起來，在發送目前消息。然後第二次發送時，因為緩存裡已經有了該消息，是以就不必再從broker拿了，且後續定時線程也會自動去更新該topic下的queue的count了。好，producer有了topic的queue的count，那使用者在發送消息時，架構就能把這個topic的queuecount傳遞給使用者，然後使用者就能根據自己的需要将消息路由到第幾個queue了。
consumer負載均衡的意思是指，在消費者叢集消費的情況下，如何讓同一個consumer group裡的消費者平均消費同一個topic下的queue。是以這個負載均衡本質上是一個将queue平均配置設定給consumer的過程。那麼怎麼實作呢？通過上面負載均衡的定義，我們隻要，要做負載均衡，必須要确定consumer group和topic；然後拿到consumer group下的所有consumer，以及topic下的所有queue；然後對于目前的consumer，就能計算出來目前consumer應該被配置設定到哪些queue了。我們可以通過如下的函數來得到目前的consumer應該被配置設定到哪幾個queue。
函數裡的實作就不多分析了。這個函數的目的就是根據給定的輸入，傳回目前consumer該配置設定到的queue。配置設定的原則就是平均配置設定。好了，有了這個函數，我們就能很友善的實作負載均衡了。我們可以對每一個正在運作的consumer内部開一個定時job，該job每隔一段時間進行一次負載均衡，也就是執行一次上面的函數，得到目前consumer該綁定的最新queue。因為每個consumer都有一個groupname屬性，用于表示目前consumer屬于哪個group。是以，我們就可以在負載均衡時到broker擷取目前group下的所有consumer；另一方面，因為每個consumer都知道它自己訂閱了哪些topic，是以有了topic資訊，就能擷取topic下的所有queue的資訊了，有了這兩樣資訊，每個consumer就能自己做負載均衡了。先看一下下面的代碼：
每個consumer内部都會啟動三個定時的task，第一個task表示要定時做一次負載均衡；第二個task表示要定時更新目前consumer訂閱的所有topic的queuecount資訊，并把最新的queuecount資訊都儲存在本地；第三個task表示目前consumer會向broker定時發送心跳，這樣broker就能通過心跳知道某個consumer是否還活着，broker上維護了所有的consumer資訊。一旦有新增或者發現沒有及時發送心跳過來的consumer，就會認為有新增或者死掉的consumer。因為broker上維護了所有的consumer資訊，是以他就能提供查詢服務，比如根據某個consumer group查詢該group下的consumer。
通過這三個定時任務，就能完成消費者的負載均衡了。先看一下rebalance方法：
代碼很簡單，就是對每個訂閱的topic做負載均衡處理。再看一下rebalanceclustering方法：
view code
上面的代碼不多分析了，就是先根據consumer group和topic擷取所有的consumer，然後對consumer做排序處理。之是以要做排序處理是為了確定負載均衡時對已有的配置設定情況盡量不發生改變。接下來就是從本地擷取topic下的所有queue，同樣根據queueid做一下排序。然後就是調用上面的配置設定算法計算出目前consumer應該配置設定到哪些queue。最後調用updatepullrequestdict方法，用來對新增或删除的queue做處理。對于新增的queue，要建立一個獨立的worker線程，開始從broker拉取消息；對于删除的queue，要停止其對應的work，停止拉取消息。
通過上面的介紹和分析，我們大家知道了equeue是如何實作消費者的負載均衡的。我們可以看出，因為每個topic下的queue的更新是異步的定時的，且負載均衡本身也是定時的，且broker上維護的consumer的資訊也不是事實的，因為每個consumer發送心跳到broker不是實時發送的，而是比如每隔5s發送一次。所有這些因為都是異步的設計，是以可能會導緻在負載均衡的過程中，同一個queue可能會被兩個消費者同時消費。這個就是所謂的，我們隻能做到一個消息至少被消費一次，但equeue層面做不到一個消息隻會被消費一次。實際上像rocketmq這種也是這樣的思路，放棄一個消息隻會被消費一次的實作（因為代價太大，且過于複雜，實際上對于分布式的環境，不太可能做到一個消息隻會被消費一次），而是采用確定一個消息至少會被消費一次（即at least once）.是以使用equeue，應用方要自己做好對每個消息的幂等處理。
消息的實時推送，一般有兩種做法：推模式（push）和拉模式（pull）。push的方式是指broker主動對所有訂閱了該topic的消費者推送消息；pull的方式是指消費者主動到broker上拉取消息；對于推模式，最大的好處就是實時，因為一有新的消息，就會立即推送給消費者。但是有一個缺點就是如果消費者來不及消費，它也會給消費者推消息，這樣就會導緻消費者端的消息會堵塞。而通過拉的方式，有兩種實作：1）輪訓的方式拉，比如每隔5s輪訓一下是否有新消息，這種方式的缺點是消息不實時，但是消費進度完全由消費者自己把控了；2）開長連接配接的方式來拉，就是不輪訓，消費者和broker之間一直保持的連接配接通道，然後broker一有新消息，就會利用這個通道把消息發送給消費者。
equeue中目前采用的是通過長連接配接拉取消息的方式。長連接配接通過socket長連接配接實作。但是雖然叫長連接配接，也不是一直不斷開，而是也會設計一個逾時的限制，比如一個長連接配接最大不超過15s，超過15s，則broker發送回複給consumer，告訴consumer目前沒有新消息；然後consumer接受到這個回複後，就知道要繼續發起下一個長連接配接來拉取。然後假如在這15s中之内，broker上有新消息了，則broker就能立即主動利用這個長連接配接通知相應的消費者，把消息傳給消費者。是以，可以看出，broker上在處理消費者的拉取消息的請求時，如果目前沒有新消息，則會hold住這個socket連接配接，最多hold 15s，超過15s，則發送傳回資訊，告訴消費者目前無消息，然後消費者再次發送pull message request過來。通過這樣的基于長連接配接的拉取模式，我們可以實作兩個好處：1）消息實時推送；2）由消費者控制消息消費進度；
另外，equeue裡還實作了消費者自身的自動限流功能。就是假如目前broker上消息很多，即生産者生産消息的速度大于消費者消費消息的速度，那broker上就會有消息被堆積。那此時消費者在拉取消息時，總是會有新消息拉取到，但是消費者又來不及處理這麼多消息。是以equeue架構内置了一個限流（流控，流量控制）的設計，就是可以允許用于配制一個消費者端堆積的消息的上限，比如3000，超過這個數目（可配置），則equeue會讓消費者以慢一點的頻率拉取消息。比如延遲個多少毫秒（延遲時間可配置）再拉取。這樣就簡單的實作了流控的目的。
作為一個消息隊列，消費者總是可能會在消費消息時抛出異常，在equeue中這種情況就是消息消費失敗的情況。通過上面的消費進度的介紹，大家知道了每個queue對某個特定的consumer group，都有一個唯一的消費進度。實際上，消息被拉取到consumer本地後，可能會被以兩種方式消費，一種是并行消費，一種是線性消費。
并行消費的意思是，假如目前一次性拉取過來32個消息，那equeue會通過啟動task（即開多線程）的方式并行消費每個消息；
線性消費的意思是，消息是在一個獨立的單線程中順序消費，消費順序和拉取過來的順序相同。
對于線性消費，假如前一個消息消費的時候失敗了，也就是抛異常了，那該怎麼辦呢？可能想到的辦法是重試個3次，但是要是重試後還是失敗呢？總不能因為這個消息而導緻後面的消息無法把消費吧？呵呵！對于這種情況，先說一下rocketmq裡的處理方式吧：它的做法是，當遇到消費失敗的情況，沒有立馬重試，而是直接把這個消息發送到broker上的某個重試隊列，發送成功後，就可以往下消費下一個消息了。因為一旦發送到重試隊列，那意味着這個消息就最後總是會被消費了，因為該消息不會丢了。但是要是發送到broker的重試隊列也不成功呢？這個？！其實這種情況不大應該出現，如果出現，那基本就是broker挂了，呵呵。
rocketmq中，對于這種情況，那會把這個失敗的消息放入本地記憶體隊列，慢慢消費它。然後繼續往後消費後面的消息。現在你一定很關心queue的offset是如何更新的？這裡涉及到一個滑動門的概念。當一批消息從broker拉取到消費者本地後，并不是馬上消費的，而是先放入一個本地的sorteddictionary，key就是消息在queue裡的位置，value就是消息本身。因為是一個排序的dictionary，是以key最小的消息意味着是最前面的消息，最大的消息就是最後面的消息。然後不管是并行消費還是線性消費，隻要某個消息被消費了，那就從這個sorteddictionary裡移除掉。每次被移除一個消息時，總是會傳回目前這個sorteddictionary裡的最小的key，然後我們就能判斷這個key是否和上次比是否前移了，如果是，則更新queue的這個最新的offset。因為每次移除一個消息的時候，總是傳回目前sorteddictionary裡的最小的key，是以，假如目前offset是3，然後offset為4的這個消息一直消費失敗，是以不會被移除，但是offset為5,6,7,8的這些消息雖然都消費成功了，但是隻要offset為4的這個消息沒有被移除，那最小的key就不會往前移動。這個就是所謂的滑動門的概念了。就好比是在鐵軌上一輛在跑的動車，offset的往前移動就好比是動車在不斷往前移動。因為我們希望offset總是會不斷往前移動，是以不希望前面的某個消費失敗的消息讓這個滑動門停止移動（即我們總是希望這個最小的key能不斷變大），是以我們會想方設法讓消費失敗的消息能不阻礙滑動門的往前移動。是以才把消費失敗的消息放入重試隊列。
另外一點需要注意一下：并不是每次成功消費完一個消息，就會立馬告訴broker更新offset，因為這樣那性能肯定很低，broker也會忙死，更好的辦法是先隻是在本地記憶體更新queue的offset，然後定時比如5s一次，将最新的offset更新到broker。是以，因為這個異步的存在，同樣也會導緻某個消息被重複消費的可能性，因為broker上的offset肯定比實際的消費進度要慢，有5s的時間差。是以，再次強調，應用方必須要處理好對消息的幂等處理！比如enode架構中，對每個command消息，架構内部都做了command的幂等處理。是以使用enode架構的應用，自身無需對command做幂等處理方面的考慮。
上面提到了并行消費和線性消費，其實對于offset的更新來說是一樣的，因為并行消費無非是多線程同時從sorteddictionary中移除消費成功的消息，而單線程隻是單個線程去移除sorteddictionary中的消息。是以我們要通過鎖的機制，保證對sorteddictionary的操作是線程安全的。目前用了readerwriterlockslim來實作對方法調用的線層安全。有興趣的朋友可以去看一下代碼。
最後，也是重點，呵呵。equeue目前還沒有實作将失敗的消息發回到broker的重試隊列。這個功能以後會考慮加進去。
這個問題比較複雜，目前equeue不支援broker的master-salve或master-master，而是單點的。我覺得一個成熟的消息隊列，為了確定在一個broker挂了的時候，要盡量能確定有其他broker可以接替它，這樣才能讓消息隊列伺服器的可靠性。但是這個問題實在太複雜。rocketmq目前實作的也隻是master-slave的方式。也就是隻要主的master挂了，那producer就無法向broker發送消息了，因為slave的broker是隻讀的，不能直接接受新消息，slave的broker隻能允許被consumer拉取消息。
這個問題，要讨論清楚，需要很多分布式方面的知識。由于篇幅的原因，這裡就不做讨論了，實際上我自己也搞不清楚到底該如何設計。希望大牛們多多指點，如何實作broker的高可用哈！