java開發RocketMQ生產者高可用示例詳解
引言
前邊兩章說瞭點基礎的,從這章開始,我們挖挖源碼。看看RocketMQ是怎麼工作的。
首先呢,這個生產者就是送孩子去碼頭的傢長,孩子們呢,就是消息瞭。
我們看看消息孩子們都長啥樣。
1 消息
public class Message implements Serializable { private static final long serialVersionUID = 8445773977080406428L; //主題名字 private String topic; //消息擴展信息,Tag,keys,延遲級別都存在這裡 private Map<String, String> properties; //消息體,字節數組 private byte[] body; //設置消息的key, public void setKeys(String keys) {} //設置topic public void setTopic(String topic) {} //延遲級別 public int setDelayTimeLevel(int level) {} //消息過濾的標記 public void setTags(String tags) {} //擴展信息存放在此 public void putUserProperty(final String name, final String value) {} }
消息就是孩子們,這些孩子們呢,有各自的特點,也有共性。同一個傢長送來的兩個孩子可以是去同一個地方的,也可以是去不同的地方的。
1.1 topic
首先呢,每個孩子消息都有一個屬性topic,這個我們上文說到瞭,是一個候船大廳。孩子們進來之後,走到自己指定的候船大廳的指定區域(平時出門坐火車高鐵不也是指定的站臺乘車麼),坐到message queue座位上等,等著出行。
Broker有一個或者多個topic,消息會存放到topic內的message queue內,等待被消費。
1.2 Body
孩子消息,也有一個Body屬性,這就是他的能力,他會畫畫,他會唱歌,他會幹啥幹啥,就記錄在這個Body屬性裡。等走出去瞭,體現價值的地方也是這個Body屬性。
Body就是消息體,消費者會根據消息體執行對應的操作。
1.3 tag
這個tag我們上節說瞭,就是一個標記,有的孩子背著畫板,相機,有的遊船就特意找到這些孩子拉走,完成他們的任務。
可以給消息設置tag屬性,消費者可以選擇含有特定tag屬性的消息進行消費。
1.4 key
key就是每個孩子消息的名字瞭。要找哪個孩子,喊他名就行。
對發送的消息設置好 Key,以後可以根據這個Key 來查找消息。比如消息異常,消息丟失,進行查找會很方便。
1.5 延遲級別
當然,還有的孩子來就不急著走,來之前就想好瞭,要恰個飯,得30分鐘,所以自己來瞭會等30分鐘後被接走。
設置延遲級別可以規定多久後消息可以被消費。
2 生產者高可用
每個送孩子來的傢長都希望能送到候船大廳裡,更不希望孩子被搞丟瞭,這個時候這個候船大廳就需要一些保證機制瞭。
2.1 客戶端保證生產者高可用
2.1.1 重試機制
就是說傢長送來瞭,孩子進到候船大廳之後,沒能成功坐到message queue座位上,這個時候工作人員會安排重試,再去看是否有座位坐。重試次數默認是2次,也就是說,消息孩子共有3次找座位坐的機會。
看源碼,我特意加瞭註解,大致可以看懂一些瞭。
//這裡取到瞭重試的次數 int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1; int times = 0; String[] brokersSent = new String[timesTotal]; for (; times < timesTotal; times++) { String lastBrokerName = null == mq ? null : mq.getBrokerName(); //獲取消息隊列 MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName); if (mqSelected != null) { mq = mqSelected; brokersSent[times] = mq.getBrokerName(); try { beginTimestampPrev = System.currentTimeMillis(); if (times > 0) { //Reset topic with namespace during resend. msg.setTopic(this.defaultMQProducer.withNamespace(msg.getTopic())); } long costTime = beginTimestampPrev - beginTimestampFirst; if (timeout < costTime) { callTimeout = true; break; } //發送消息 sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime); ... } catch (RemotingException e) { ... continue; } catch (MQClientException e) { ... continue; } catch (MQBrokerException e) { ... continue; } catch (InterruptedException e) { //可以看到隻有InterruptedException拋出瞭異常,其他的exception都會繼續重試 throw e; } } else { break; } }
重試代碼如上,這個sendDefaultImpl
方法中,會嘗試發送三次消息,若是都失敗,才會拋出對應的錯誤。
2.1.2 客戶端容錯
若是有多個Broker候車大廳的時候,服務人員會安排消息孩子選擇一個相對不擁擠,比較容易進入的來進入。當然那些已經關閉的,停電的,沒有服務能力的,我們是不會進的。
MQ Client會維護一個Broker的發送延遲信息,根據這個信息會選擇一個相對延遲較低的Broker來發送消息。會主動剔除哪些已經宕機,不可用或發送延遲級別較高的Broker.
選擇Broker
就是在選擇message queue
,對應的代碼如下:
這裡會先判斷延遲容錯開關是否開啟,這個開關默認是關閉的,若是開啟的話,會優先選擇延遲較低的Broker。
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) { //判斷發送延遲容錯開關是否開啟 if (this.sendLatencyFaultEnable) { try { //選擇一個延遲上可以接受,並且和上次發送相同的Broker int index = tpInfo.getSendWhichQueue().incrementAndGet(); for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) { int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size(); if (pos < 0) pos = 0; MessageQueue mq = tpInfo.getMessageQueueList().get(pos); //若是Broker的延遲時間可以接受,則返回這個Broker if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) return mq; } //若是第一步沒能選中一個Broker,就選擇一個延遲較低的Broker final String notBestBroker = latencyFaultTolerance.pickOneAtLeast(); int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker); if (writeQueueNums > 0) { final MessageQueue mq = tpInfo.selectOneMessageQueue(); if (notBestBroker != null) { mq.setBrokerName(notBestBroker); mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums); } return mq; } else { latencyFaultTolerance.remove(notBestBroker); } } catch (Exception e) { log.error("Error occurred when selecting message queue", e); } //若是前邊都沒選中一個Broker,就隨機選一個Broker return tpInfo.selectOneMessageQueue(); } return tpInfo.selectOneMessageQueue(lastBrokerName); }
但是當延遲容錯開關為關閉狀態的時候,執行的代碼如下:
為瞭均勻分散Broker的壓力,會選擇與之前不同的Broker。
public MessageQueue selectOneMessageQueue(final String lastBrokerName) { //若是沒有上次的Brokername做參考,就隨機選一個 if (lastBrokerName == null) { return selectOneMessageQueue(); } else { //如果有,那麼就選一個其他的Broker for (int i = 0; i < this.messageQueueList.size(); i++) { int index = this.sendWhichQueue.incrementAndGet(); int pos = Math.abs(index) % this.messageQueueList.size(); if (pos < 0) pos = 0; MessageQueue mq = this.messageQueueList.get(pos); //這裡判斷遇上一個使用的Broker不是同一個 if (!mq.getBrokerName().equals(lastBrokerName)) { return mq; } } //若是上邊的都沒選中,那麼就隨機選一個 return selectOneMessageQueue(); } }
2.2 Broker端保證生產者高可用
Broker候船大廳為瞭能確切的接收到消息孩子,至少會有兩個廳,一個主廳一個副廳,一般來說孩子都會進入到主廳,然後一頓操作,卡該忙信那機資(影分身之術),然後讓分身進入到副廳,這樣當主廳停電瞭,不工作瞭,副廳的分身隻要去完成瞭任務就ok的。一般來說都是主廳的消息孩子去坐船完成任務。
之後我們會聊到Broker的主從復制,分為同步復制和異步復制,同步復制時指當master 收到消息之後,同步到slaver才算消息發送成功。異步復制是隻要master收到消息就算成功。生產中建議至少部署兩臺master和兩臺slaver。
下一篇,我們聊聊,消息的發送流程,就是說,一個消息孩子,從進碼頭的門到坐到message queue座位上,都經歷瞭啥。
以上就是java開發RocketMQ生產者高可用示例詳解的詳細內容,更多關於java RocketMQ生產者高可用的資料請關註WalkonNet其它相關文章!
推薦閱讀:
- RocketMQ消息發送流程源碼剖析
- 分佈式消息隊列RocketMQ概念詳解
- MQ的分類組成優缺點測試點入門教程
- RocketMQ Namesrv架構工作原理詳解
- 一文徹底掌握RocketMQ 的存儲模型