RocketMQ系列(一)基本概念


RocketMQ是阿里出品的一款開源的消息中間件,讓其聲名大噪的就是它的事務消息的功能。在企業中,消息中間件選擇使用RocketMQ的還是挺多的,這一系列的文章都是針對RocketMQ的,咱們先從RocketMQ的一些基本概念和環境的搭建開始聊起。

RocketMQ由4部分組成,分別是:名稱服務(Name Server)、消息隊列(Brokers)、生產者(producer)和消費者(consumer)。這4部分都可以進行水平擴展,從而避免單點故障,如下圖,

這是RocketMQ官網上的一張圖,非常清晰的列出了4個部分,並且都是集群模式。下面我們就分別說一說這4部分。

名稱服務(NameServer)

Name Server扮演的角色是一個注冊中心,和Zookeeper的作用差不多。它的主要功能有兩個,如下:

  • broker的管理:broker集群將自己的信息注冊到NameServer,NameServer提供心跳機制檢測每一個broker是否正常。
  • 路由管理:每一個NameServer都有整個broker集群和隊列的信息,以便客戶端(生產者和消費者)查詢。

NameServer協調着分布式系統中的每一個組件,並且管理着每一個Topic的路由信息。

Broker

Broker主要是存儲消息,並且提供Topic的機制。它提供推和拉兩種模式,還有一些容災的措施,比如可以配置消息副本。下面我們看一看Brokcer的主從機制。

Broker的角色分為“異步主”、“同步主”和“從”三個角色。如果你不能容忍消息的丟失,你可以配置一個“同步主”和“從”兩個Broker,如果你覺得消息丟失也無所謂,只要隊列可用就ok的話,你可以配置“異步主”和“從”兩個broker。如果你只是想簡單的搭建,只配置一個“異步主”,不配置“從”也是可以的。

上面提到的是broker之間的備份,broker里的信息也是可以保存到磁盤的,保存到磁盤的方式也有兩種,推薦的方式是異步保存磁盤,同步保存磁盤是非常損耗性能的。

生產者

生產者支持集群部署,它們向broker集群發送消息,而且支持多種負載均衡的方式。

當生產者向broker發送消息時,會得到發送結果,發送結果中有一個發送狀態。假設我們的配置中,消息的配置isWaitStoreMsgOK = true,這個配置默認也是true,如果你配置為false,在發送消息的過程中,只要不發生異常,發送結果都是SEND_OK。當isWaitStoreMsgOK = true,發送結果有以下幾種,

  • FLUSH_DISK_TIMEOUT:保存磁盤超時,當保存磁盤的方式設置為SYNC_FLUSH(同步),並且在syncFlushTimeout配置的時間內(默認5s),沒有完成保存磁盤的動作,將會得到這個狀態。
  • FLUSH_SLAVE_TIMEOUT:同步“從”超時,當broker的角色設置為“同步主”時,但是在設置的同步時間內,默認為5s,沒有完成主從之間的同步,就會得到這個狀態。
  • SLAVE_NOT_AVAILABLE:“從”不可用,當我們設置“同步主”,但是沒有配置“從”broker時,會返回這個狀態。
  • SEND_OK:消息發送成功。

再來看看消息重復與消息丟失,當你發現你的消息丟失時,通常有兩個選擇,一個是丟就丟吧,這樣消息就真的丟了;另一個選擇是消息重新發送,這樣有可能引起消息重復。通常情況下,還是推薦重新發送的,我們在消費消息的時候要去除掉重復的消息。

發送message的大小一般不超過512k,默認的發送消息的方式是同步的,發送方法會一直阻塞,直到等到返回的響應。如果你比較在意性能,也可以用send(msg, callback)異步的方式發送消息。

消費者

多個消費者可以組成消費者組(consumer group),不同的消費者組可以訂閱相同的Topic,也可以獨立的消費Topic,每一個消費者組都有自己的消費偏移量。

消息的消費方式一般有兩種,順序消費和並發消費。

  • 順序消費:消費者將鎖住消息隊列,確保消息按照順序一個一個的被消費掉,順序消費會引起一部分性能損失。在消費消息的時候,如果出現異常,不建議直接拋出,而是應該返回SUSPEND_CURRENT_QUEUE_A_MOMENT 這個狀態,它將告訴消費者過一段時間后,會重新消費這個消息。
  • 並發消費:消費者將並發的消費消息,這種方式的性能非常好,也是推薦的消費方式。在消費的過程中,如果出現異常,不建議直接拋出,而是返回RECONSUME_LATER 狀態,它告訴消費者現在不能正確的消費它,過一段時間后,會再次消費它。

在消費者內部,是使用ThreadPoolExecutor作為線程池的,我們可以通過setConsumeThreadMin setConsumeThreadMax 設置最小消費線程和最大消費線程。

當一個新的消費者組建立以后,它要決定是否消費之前的歷史消息,CONSUME_FROM_LAST_OFFSET將忽略歷史消息,消費新的消息。CONSUME_FROM_FIRST_OFFSET將消費隊列中的每一個消息,之前的歷史消息也會再消費一遍。CONSUME_FROM_TIMESTAMP可以指定消費消息的時間,指定時間以后的消息會被消費。

如果你的應用不能容忍重復消費,那么在消費消息的過程中,要做好消息的校驗。

好了,今天就到這里吧,下一篇我們將介紹RocketMQ的環境搭建。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM