java開發RocketMQ生產者高可用範例詳解

2022-08-05 22:01:01

引言

前邊兩章說了點基礎的，從這章開始，我們挖挖原始碼。看看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();
    }
}