首先來看一個RcoketMQ傳送訊息的例子：
@Service
public class MQService {
@Autowired
DefaultMQProducer defaultMQProducer;
public void sendMsg() {
String msg = "我是一條訊息";
// 建立訊息，指定TOPIC、TAG和訊息內容
Message sendMsg = new Message("TestTopic", "TestTag", msg.getBytes());
SendResult sendResult = null;
try {
// 同步傳送訊息
sendResult = defaultMQProducer.send(sendMsg);
System.out.println("訊息傳送響應：" + sendResult.toString());
} catch (MQClientException e) {
e.printStackTrace();
} catch (RemotingException e) {
e.printStackTrace();
} catch (MQBrokerException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
RocketMQ是通過DefaultMQProducer進行訊息傳送的，它實現了MQProducer介面，MQProducer介面中定義了訊息傳送的方法，方法主要分為三大類：
public interface MQProducer extends MQAdmin {
// 同步傳送訊息
SendResult send(final Message msg) throws MQClientException, RemotingException, MQBrokerException,
InterruptedException;
// 非同步傳送訊息，SendCallback為回撥函數
void send(final Message msg, final SendCallback sendCallback) throws MQClientException,
RemotingException, InterruptedException;
// 非同步傳送訊息，沒有回撥函數
void sendOneway(final Message msg) throws MQClientException, RemotingException,
InterruptedException;
// 省略其他方法
}
接下來以將以同步訊息傳送為例來分析訊息傳送的流程。
DefaultMQProducer裡面有一個DefaultMQProducerImpl型別的成員變數defaultMQProducerImpl，從預設的無參建構函式中可以看出在建構函式中對defaultMQProducerImpl進行了範例化，在send方法中就是呼叫defaultMQProducerImpl的方法進行訊息傳送的：
public class DefaultMQProducer extends ClientConfig implements MQProducer {
/**
* 預設訊息生產者實現類
*/
protected final transient DefaultMQProducerImpl defaultMQProducerImpl;
/**
* 預設的建構函式
*/
public DefaultMQProducer() {
this(null, MixAll.DEFAULT_PRODUCER_GROUP, null);
}
/**
* 建構函式
*/
public DefaultMQProducer(final String namespace, final String producerGroup, RPCHook rpcHook) {
this.namespace = namespace;
this.producerGroup = producerGroup;
// 範例化
defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
}
/**
* 同步傳送訊息
*/
@Override
public SendResult send(
Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
// 設定主題
msg.setTopic(withNamespace(msg.getTopic()));
// 傳送訊息
return this.defaultMQProducerImpl.send(msg);
}
}
DefaultMQProducerImpl中訊息的傳送在sendDefaultImpl方法中實現，處理邏輯如下：
TopicPublishInfoselectOneMessageQueue方法選擇一個訊息佇列，並記錄本次選擇的訊息佇列，在下一次傳送訊息時選擇佇列時使用sendKernelImpl方法進行訊息傳送updateFaultItem記錄向Broker傳送訊息的耗時，在開啟故障延遲處理機制時使用public class DefaultMQProducerImpl implements MQProducerInner {
/**
* DEFAULT SYNC -------------------------------------------------------
*/
public SendResult send(
Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
// 傳送訊息
return send(msg, this.defaultMQProducer.getSendMsgTimeout());
}
public SendResult send(Message msg,
long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
// 傳送訊息
return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
}
/**
* 傳送訊息
* @param msg 傳送的訊息
* @param communicationMode
* @param sendCallback 回撥函數
* @param timeout 超時時間
*/
private SendResult sendDefaultImpl(
Message msg,
final CommunicationMode communicationMode,
final SendCallback sendCallback,
final long timeout
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
this.makeSureStateOK();
Validators.checkMessage(msg, this.defaultMQProducer);
final long invokeID = random.nextLong();
// 開始時間
long beginTimestampFirst = System.currentTimeMillis();
long beginTimestampPrev = beginTimestampFirst;
long endTimestamp = beginTimestampFirst;
// 查詢主題路由資訊
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
if (topicPublishInfo != null && topicPublishInfo.ok()) {
boolean callTimeout = false;
// 訊息佇列
MessageQueue mq = null;
Exception exception = null;
SendResult sendResult = null;
// 獲取失敗重試次數
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
int times = 0;
String[] brokersSent = new String[timesTotal];
for (; times < timesTotal; times++) {
// 獲取BrokerName
String lastBrokerName = null == mq ? null : mq.getBrokerName();
// 根據BrokerName選擇一個訊息佇列
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);
// 結束時間
endTimestamp = System.currentTimeMillis();
// 記錄向Broker傳送訊息的請求耗時,訊息傳送結束時間 - 開始時間
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
switch (communicationMode) {
case ASYNC:
return null;
case ONEWAY:
return null;
case SYNC:
// 如果傳送失敗
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
// 是否重試
if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
continue;
}
}
// 返回結果
return sendResult;
default:
break;
}
} catch (RemotingException e) {
endTimestamp = System.currentTimeMillis();
// 如果丟擲異常，記錄請求耗時
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
exception = e;
continue;
}
// ... 省略其他例外處理
} else {
break;
}
}
if (sendResult != null) {
return sendResult;
}
// ...
}
validateNameServerSetting();
throw new MQClientException("No route info of this topic: " + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),
null).setResponseCode(ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION);
}
}
DefaultMQProducerImpl中有一個路由資訊表topicPublishInfoTable，記錄了主題對應的路由資訊，其中KEY為topic, value為對應的路由資訊物件TopicPublishInfo：
public class DefaultMQProducerImpl implements MQProducerInner {
// 路由資訊表，KEY為topic, value為對應的路由資訊物件TopicPublishInfo
private final ConcurrentMap<String, TopicPublishInfo> topicPublishInfoTable =
new ConcurrentHashMap<String, TopicPublishInfo>();
}
TopicPublishInfo中記錄了主題所在的訊息佇列資訊、所在Broker等資訊：
messageQueueList：一個MessageQueue型別的訊息佇列列表，MessageQueue中記錄了主題名稱、主題所屬的Broker名稱和佇列ID
sendWhichQueue：計數器，選擇訊息佇列的時候增1，以此達到輪詢的目的
topicRouteData：從NameServer查詢到的主題對應的路由資料，包含了佇列和Broker的相關資料
public class TopicPublishInfo {
// 訊息佇列列表
private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>();
// 一個計數器，每次選擇訊息佇列的時候增1，以此達到輪詢的目的
private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
// 主題路由資料
private TopicRouteData topicRouteData;
// ...
}
// 訊息佇列
public class MessageQueue implements Comparable<MessageQueue>, Serializable {
private static final long serialVersionUID = 6191200464116433425L;
private String topic; // 主題
private String brokerName; // 所屬Broker名稱
private int queueId; // 佇列ID
// ...
}
// 主題路由資料
public class TopicRouteData extends RemotingSerializable {
private List<QueueData> queueDatas; // 佇列資料列表
private List<BrokerData> brokerDatas; // Broker資訊列表
// ...
}
// 佇列資料
public class QueueData implements Comparable<QueueData> {
private String brokerName; // Broker名稱
private int readQueueNums; // 可讀佇列數量
private int writeQueueNums; // 可寫佇列數量
private int perm;
private int topicSysFlag;
}
// Broker資料
public class BrokerData implements Comparable<BrokerData> {
private String cluster; // 叢集名稱
private String brokerName; // Broker名稱
private HashMap<Long, String> brokerAddrs; // Broker地址集合，KEY為Broker ID, value為Broker 地址
// ...
}
在查詢主題路由資訊的時候首先從DefaultMQProducerImpl快取的路由表topicPublishInfoTable中根據主題查詢路由資訊，如果查詢成功返回即可，如果未查詢到，需要從NameServer中獲取路由資訊，如果獲取失敗，則使用預設的主題路由資訊：
public class DefaultMQProducerImpl implements MQProducerInner {
// 路由資訊表，KEY為topic, value為對應的路由資訊物件TopicPublishInfo
private final ConcurrentMap<String, TopicPublishInfo> topicPublishInfoTable =
new ConcurrentHashMap<String, TopicPublishInfo>();
/**
* 根據主題查詢路由資訊
* @param topic 主題
* @return
*/
private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
// 根據主題獲取對應的主題路由資訊
TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
// 如果未獲取到
if (null == topicPublishInfo || !topicPublishInfo.ok()) {
this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
// 從NameServer中查詢路由資訊
this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
topicPublishInfo = this.topicPublishInfoTable.get(topic);
}
// 如果路由資訊獲取成功
if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
// 返回路由資訊
return topicPublishInfo;
} else {
// 如果路由資訊未獲取成功，使用預設主題查詢路由資訊
this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
topicPublishInfo = this.topicPublishInfoTable.get(topic);
// 返回路由資訊
return topicPublishInfo;
}
}
}
從NameServer獲取主題路由資訊資料是在MQClientInstance中的updateTopicRouteInfoFromNameServer方法中實現的：
TopicRouteData型別的物件返回topicRouteTable主題路由表中根據主題獲取舊的路由資訊，與新的對比，判斷資訊是否發生了變化，如果傳送了變化需要更新brokerAddrTable中記錄的資料topicRouteTable中，替換掉舊的資訊public class MQClientInstance {
public boolean updateTopicRouteInfoFromNameServer(final String topic) {
// 從NameServer更新路由資訊
return updateTopicRouteInfoFromNameServer(topic, false, null);
}
/**
* 從NameServer更新路由資訊
* @param topic 主題
* @param isDefault 是否使用預設的主題
* @param defaultMQProducer 預設訊息生產者
* @return
*/
public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
DefaultMQProducer defaultMQProducer) {
try {
if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
try {
TopicRouteData topicRouteData;
// 是否使用預設的路由資訊
if (isDefault && defaultMQProducer != null) {
// 使用預設的主題路由資訊
topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),
clientConfig.getMqClientApiTimeout());
if (topicRouteData != null) {
for (QueueData data : topicRouteData.getQueueDatas()) {
int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
data.setReadQueueNums(queueNums); // 設定可讀佇列數量
data.setWriteQueueNums(queueNums); // 設定可寫佇列數量
}
}
} else {
// 從NameServer獲取路由資訊
topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, clientConfig.getMqClientApiTimeout());
}
// 如果路由資訊不為空
if (topicRouteData != null) {
// 從路由表中獲取舊的路由資訊
TopicRouteData old = this.topicRouteTable.get(topic);
// 判斷路由資訊是否發生變化
boolean changed = topicRouteDataIsChange(old, topicRouteData);
if (!changed) {
// 是否需要更新路由資訊
changed = this.isNeedUpdateTopicRouteInfo(topic);
} else {
log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
}
// 如果資料發生變化
if (changed) {
// 克隆一份新的路由資訊
TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();
// 處理brokerAddrTable中的資料
for (BrokerData bd : topicRouteData.getBrokerDatas()) {
// 更新brokerAddrTable中的資料
this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
}
// ...
log.info("topicRouteTable.put. Topic = {}, TopicRouteData[{}]", topic, cloneTopicRouteData);
// 將新的路由資訊加入到路由表
this.topicRouteTable.put(topic, cloneTopicRouteData);
return true;
}
} else {
log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}. [{}]", topic, this.clientId);
}
} catch (MQClientException e) {
if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) && !topic.equals(TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC)) {
log.warn("updateTopicRouteInfoFromNameServer Exception", e);
}
} catch (RemotingException e) {
log.error("updateTopicRouteInfoFromNameServer Exception", e);
throw new IllegalStateException(e);
} finally {
this.lockNamesrv.unlock();
}
} else {
log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms. [{}]", LOCK_TIMEOUT_MILLIS, this.clientId);
}
} catch (InterruptedException e) {
log.warn("updateTopicRouteInfoFromNameServer Exception", e);
}
return false;
}
}
向NameServer傳送請求的程式碼實現在MQClientAPIImpl的getTopicRouteInfoFromNameServer方法中，可以看到構建了請求命令RemotingCommand並設定請求型別為RequestCode.GET_ROUTEINFO_BY_TOPIC，表示從NameServer獲取路由資訊，之後通過Netty向NameServer傳送請求，並解析返回結果：
public class MQClientAPIImpl {
public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis)
throws RemotingException, MQClientException, InterruptedException {
// 從NameServer獲取路由資訊
return getTopicRouteInfoFromNameServer(topic, timeoutMillis, true);
}
/**
* 從NameServer獲取路由資訊
* @param topic
* @param timeoutMillis
* @param allowTopicNotExist
* @return
* @throws MQClientException
* @throws InterruptedException
* @throws RemotingTimeoutException
* @throws RemotingSendRequestException
* @throws RemotingConnectException
*/
public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis,
boolean allowTopicNotExist) throws MQClientException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader();
requestHeader.setTopic(topic);
// 建立請求命令，請求型別為獲取主題路由資訊GET_ROUTEINFO_BY_TOPIC
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINFO_BY_TOPIC, requestHeader);
// 傳送請求
RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);
assert response != null;
switch (response.getCode()) {
// 如果主題不存在
case ResponseCode.TOPIC_NOT_EXIST: {
if (allowTopicNotExist) {
log.warn("get Topic [{}] RouteInfoFromNameServer is not exist value", topic);
}
break;
}
// 如果請求傳送成功
case ResponseCode.SUCCESS: {
byte[] body = response.getBody();
// 返回獲取的路由資訊
if (body != null) {
return TopicRouteData.decode(body, TopicRouteData.class);
}
}
default:
break;
}
throw new MQClientException(response.getCode(), response.getRemark());
}
}
主題路由資訊資料TopicPublishInfo獲取到之後，需要從中選取一個訊息佇列，是通過呼叫MQFaultStrategy的selectOneMessageQueue方法觸發的，之後會進入MQFaultStrategy的selectOneMessageQueue方法從主題路由資訊中選擇訊息佇列：
public class DefaultMQProducerImpl implements MQProducerInner {
private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy();
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
// 選擇訊息佇列
return this.mqFaultStrategy.selectOneMessageQueue(tpInfo, lastBrokerName);
}
}
MQFaultStrategy的selectOneMessageQueue方法主要是通過呼叫TopicPublishInfo中的相關方法進行訊息佇列選擇的。
啟用故障延遲機制
如果啟用了故障延遲機制，會遍歷TopicPublishInfo中儲存的訊息佇列列表，對計數器增1，輪詢選擇一個訊息佇列，接著會判斷訊息佇列所屬的Broker是否可用，如果Broker可用返回訊息佇列即可。
如果選出的佇列所屬Broker不可用，會呼叫latencyFaultTolerance的pickOneAtLeast方法（下面會講到）選擇一個Broker，從tpInfo中獲取此Broker可寫的佇列數量，如果數量大於0，呼叫selectOneMessageQueue()方法選擇一個佇列。
如果故障延遲機制未選出訊息佇列，依舊會呼叫selectOneMessageQueue()選擇出一個訊息佇列。
未啟用故障延遲機制
直接呼叫的selectOneMessageQueue(String lastBrokerName)方法並傳入上一次使用的Broker名稱進行選擇。
public class MQFaultStrategy {
/**
* 選擇訊息佇列
* @param tpInfo 主題路由資訊
* @param lastBrokerName 上一次使用的Broker名稱
* @return
*/
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
// 如果啟用故障延遲機制
if (this.sendLatencyFaultEnable) {
try {
// 計數器增1
int index = tpInfo.getSendWhichQueue().incrementAndGet();
// 遍歷TopicPublishInfo中儲存的訊息佇列列表
for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
// 輪詢選擇一個訊息佇列
int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
// 如果下標小於0，則使用0
if (pos < 0)
pos = 0;
// 根據下標獲取訊息佇列
MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
// 判斷訊息佇列所屬的Broker是否可用,如果可用返回當前選擇的訊息佇列
if (latencyFaultTolerance.isAvailable(mq.getBrokerName()))
return mq;
}
// 如果未獲取到可用的Broker
// 呼叫pickOneAtLeast選擇一個
final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
// 從tpInfo中獲取Broker可寫的佇列數量
int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
// 如果可寫的佇列數量大於0
if (writeQueueNums > 0) {
// 選擇一個訊息佇列
final MessageQueue mq = tpInfo.selectOneMessageQueue();
if (notBestBroker != null) {
// 設定訊息佇列所屬的Broker
mq.setBrokerName(notBestBroker);
// 設定佇列ID
mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums);
}
// 返回訊息佇列
return mq;
} else {
// 移除Broker
latencyFaultTolerance.remove(notBestBroker);
}
} catch (Exception e) {
log.error("Error occurred when selecting message queue", e);
}
// 如果故障延遲機制未選出訊息佇列，呼叫selectOneMessageQueue選擇訊息佇列
return tpInfo.selectOneMessageQueue();
}
// 根據上一次使用的BrokerName獲取訊息佇列
return tpInfo.selectOneMessageQueue(lastBrokerName);
}
}
selectOneMessageQueue方法的實現
selectOneMessageQueue方法中，如果上一次選擇的BrokerName為空，則呼叫無參的selectOneMessageQueue方法選擇訊息佇列，也是預設的選擇方式，首先對計數器增一，然後用計數器的值對messageQueueList列表的長度取餘得到下標值pos，再從messageQueueList中獲取pos位置的元素，以此達到輪詢從messageQueueList列表中選擇訊息佇列的目的。
如果傳入的BrokerName不為空，遍歷messageQueueList列表，同樣對計數器增一，並對messageQueueList列表的長度取餘，選取一個訊息佇列，不同的地方是選擇訊息佇列之後，會判斷訊息佇列所屬的Broker是否與上一次選擇的Broker名稱一致，如果一致則繼續迴圈，輪詢選擇下一個訊息佇列，也就是說，如果上一次選擇了某個Broker傳送訊息，本次將不會再選擇這個Broker，當然如果最後仍未找到滿足要求的訊息佇列，則仍舊使用預設的選擇方式，也就是呼叫無參的selectOneMessageQueue方法進行選擇。
public class TopicPublishInfo {
private boolean orderTopic = false;
private boolean haveTopicRouterInfo = false;
private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>(); // 訊息佇列列表
private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex(); // 一個計數器，每次選擇訊息佇列的時候增1，以此達到輪詢的目的
private TopicRouteData topicRouteData;
// ...
public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
// 如果上一次選擇的BrokerName為空
if (lastBrokerName == null) {
// 選擇訊息佇列
return selectOneMessageQueue();
} else {
// 遍歷訊息佇列列表
for (int i = 0; i < this.messageQueueList.size(); i++) {
// 計數器增1
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();
}
}
// 選擇訊息佇列
public MessageQueue selectOneMessageQueue() {
// 自增
int index = this.sendWhichQueue.incrementAndGet();
// 對長度取餘
int pos = Math.abs(index) % this.messageQueueList.size();
if (pos < 0)
pos = 0;
// 選擇訊息佇列
return this.messageQueueList.get(pos);
}
}
回到傳送訊息的程式碼中，可以看到訊息傳送無論成功與否都會呼叫updateFaultItem方法更新失敗條目：
public class DefaultMQProducerImpl implements MQProducerInner {
private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy();
// 傳送訊息
private SendResult sendDefaultImpl(
Message msg,
final CommunicationMode communicationMode,
final SendCallback sendCallback,
final long timeout
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
// ...
for (; times < timesTotal; times++) {
try {
// 開始時間
beginTimestampPrev = System.currentTimeMillis();
// ...
// 傳送訊息
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
// 結束時間
endTimestamp = System.currentTimeMillis();
// 更新失敗條目
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
// ...
} catch (RemotingException e) {
endTimestamp = System.currentTimeMillis();
// 更新失敗條目
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
exception = e;
continue;
}
// 省略其他catch
// ...
catch (InterruptedException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);
log.warn(msg.toString());
log.warn("sendKernelImpl exception", e);
log.warn(msg.toString());
throw e;
}
} else {
break;
}
}
// ...
}
// 更新FaultItem
public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
// 呼叫MQFaultStrategy的updateFaultItem方法
this.mqFaultStrategy.updateFaultItem(brokerName, currentLatency, isolation);
}
}
MQFaultStrategy中有一個型別的成員變數，最終是通過呼叫latencyFaultTolerance的updateFaultItem方法進行更新的，並傳入了三個引數：
brokerName：Broker名稱
currentLatency：當前延遲時間，由上面的呼叫可知傳入的值為傳送訊息的耗時時間，即訊息傳送結束時間 - 開始時間
duration：持續時間，根據isolation的值決定，如果為true，duration的值為30000ms也就是30s，否則與currentLatency的值一致
public class MQFaultStrategy {
// 故障延遲機制
private final LatencyFaultTolerance<String> latencyFaultTolerance = new LatencyFaultToleranceImpl();
/**
* 更新失敗條目
* @param brokerName Broker名稱
* @param currentLatency 傳送訊息耗時：請求結束時間 - 開始時間
* @param isolation
*/
public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
if (this.sendLatencyFaultEnable) {
// 計算duration，isolation為true時使用30000，否則使用傳送訊息的耗時時間currentLatency
long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
// 更新到latencyFaultTolerance中
this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
}
}
}
LatencyFaultToleranceImpl
LatencyFaultToleranceImpl中有一個faultItemTable，記錄了每個Broker對應的FaultItem，在updateFaultItem方法中首先根據Broker名稱從faultItemTable獲取FaultItem：
FaultItem，新建FaultItem物件，設定傳入的currentLatency延遲時間（訊息傳送結束時間 - 開始時間）和開始時間即當前時間 +notAvailableDuration，notAvailableDuration值有兩種情況，值為30000毫秒或者與currentLatency的值一致FaultItem，更新FaultItem中的currentLatency延遲時間和StartTimestamp開始時間public class LatencyFaultToleranceImpl implements LatencyFaultTolerance<String> {
// FaultItem集合，Key為BrokerName，value為對應的FaultItem物件
private final ConcurrentHashMap<String, FaultItem> faultItemTable = new ConcurrentHashMap<String, FaultItem>(16);
/**
* 更新FaultItem
* @param name Broker名稱
* @param currentLatency 延遲時間，也就是傳送訊息耗時：請求結束時間 - 開始時間
* @param notAvailableDuration 不可用的持續時間，也就是上一步中的duration
*/
@Override
public void updateFaultItem(final String name, final long currentLatency, final long notAvailableDuration) {
// 獲取FaultItem
FaultItem old = this.faultItemTable.get(name);
// 如果不存在
if (null == old) {
// 新建FaultItem
final FaultItem faultItem = new FaultItem(name);
// 設定currentLatency延遲時間
faultItem.setCurrentLatency(currentLatency);
// 設定規避故障開始時間，當前時間 + 不可用的持續時間，不可用的持續時間有兩種情況：值為30000或者與currentLatency一致
faultItem.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
// 新增到faultItemTable
old = this.faultItemTable.putIfAbsent(name, faultItem);
if (old != null) {
old.setCurrentLatency(currentLatency);
old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
}
} else {
// 更新時間
old.setCurrentLatency(currentLatency);
// 更新開始時間
old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
}
}
}
FaultItem是LatencyFaultToleranceImpl的一個內部類，裡面有三個變數：
notAvailableDuration，notAvailableDuration值有兩種情況，值為30000毫秒或者與currentLatency的值一致。isAvailable方法
isAvailable方法用於開啟故障延遲機制時判斷Broker是否可用，可用判斷方式為：當前時間 - startTimestamp的值大於等於 0，如果小於0則認為不可用。
上面分析可知startTimestamp的值為新建/更新FaultItem的時間 + 不可用的時間，如果當前時間減去規避故障開始時間的值大於等於0，說明此Broker已經超過了設定的規避時間，可以重新被選擇用於傳送訊息。
compareTo方法
FaultItem還實現了Comparable，重寫了compareTo方法，在排序的時候使用，對比大小的規則如下：
呼叫isAvailable方法判斷當前物件和other的值是否相等，如果相等繼續第2步，如果不相等，說明兩個物件一個返回true一個返回false，此時優先判斷當前物件的isAvailable方法返回值是否為true：
other小，返回-1，對應當前物件為true，other物件為false的情況。other的isAvailable方法判斷是否為true，如果為true，返回1，表示other比較大（對應當前物件為false，other物件為true的情況），否則繼續第2步根據其他條件判斷。對比currentLatency的值，如果currentLatency值小於other的，返回-1，表示當前物件比other小。
對比startTimestamp的值，如果startTimestamp值小於other的，返回-1，同樣表示當前物件比other小。
總結
isAvailable方法返回true的時候表示FaultItem物件的值越小，因為true代表Broker已經過了規避故障的時間，可以重新被選擇。
currentLatency的值越小表示FaultItem的值越小。currentLatency的值與Broker傳送訊息的耗時有關，耗時越低，值就越小。
startTimestamp值越小同樣表示整個FaultItem的值也越小。startTimestamp的值與currentLatency有關（值不為預設的30000毫秒情況下），currentLatency值越小，startTimestamp的值也越小。
public class LatencyFaultToleranceImpl implements LatencyFaultTolerance<String> {
class FaultItem implements Comparable<FaultItem> {
private final String name; // Broker名稱
private volatile long currentLatency; // 傳送訊息耗時時間：請求結束時間 - 開始時間
private volatile long startTimestamp; // 規避開始時間：新建/更新FaultItem的時間 + 不可用的時間notAvailableDuration
@Override
public int compareTo(final FaultItem other) {
// 如果isAvailable不相等，說明一個為true一個為false
if (this.isAvailable() != other.isAvailable()) {
if (this.isAvailable()) // 如果當前物件為true
return -1; // 當前物件小
if (other.isAvailable())// 如果other物件為true
return 1; // other物件大
}
// 對比傳送訊息耗時時間
if (this.currentLatency < other.currentLatency)
return -1;// 當前物件小
else if (this.currentLatency > other.currentLatency) {
return 1; // other物件大
}
// 對比故障規避開始時間
if (this.startTimestamp < other.startTimestamp)
return -1;
else if (this.startTimestamp > other.startTimestamp) {
return 1;
}
return 0;
}
// 用於判斷Broker是否可用
public boolean isAvailable() {
// 當前時間減去startTimestamp的值是否大於等於0，大於等於0表示可用
return (System.currentTimeMillis() - startTimestamp) >= 0;
}
}
}
在選擇訊息佇列時，如果開啟故障延遲機制並且未找到合適的訊息佇列，會呼叫pickOneAtLeast方法選擇一個Broker，那麼是如何選擇Broker的呢？
首先遍歷faultItemTableMap集合，將每一個Broker對應的FaultItem加入到LinkedList連結串列中
呼叫sort方法對連結串列進行排序，預設是正序從小到大排序，FaultItem還實現Comparable就是為了在這裡進行排序，值小的排在連結串列前面
計算中間值half：
half值小於等於0，取連結串列中的第一個元素half值大於0，從前half個元素中輪詢選擇元素由FaultItem的compareTo方法可知，currentLatency和startTimestamp的值越小，整個FaultItem的值也就越小，正序排序時越靠前，靠前表示向Broker傳送訊息的延遲越低，在選擇Broker時優先順序越高，所以如果half值小於等於0的時候，取連結串列中的第一個元素，half值大於0的時候，處於連結串列前half個的Brokerddd，延遲都是相對較低的，此時輪詢從前haft個Broker中選擇一個Broker。
public class LatencyFaultToleranceImpl implements LatencyFaultTolerance<String> {
// FaultItem集合，Key為BrokerName，value為對應的FaultItem物件
private final ConcurrentHashMap<String, FaultItem> faultItemTable = new ConcurrentHashMap<String, FaultItem>(16);
@Override
public String pickOneAtLeast() {
final Enumeration<FaultItem> elements = this.faultItemTable.elements();
List<FaultItem> tmpList = new LinkedList<FaultItem>();
// 遍歷faultItemTable
while (elements.hasMoreElements()) {
final FaultItem faultItem = elements.nextElement();
// 將FaultItem新增到列表中
tmpList.add(faultItem);
}
if (!tmpList.isEmpty()) {
Collections.shuffle(tmpList);
// 排序
Collections.sort(tmpList);
// 計算中間數
final int half = tmpList.size() / 2;
// 如果中位數小於等於0
if (half <= 0) {
// 獲取第一個元素
return tmpList.get(0).getName();
} else {
// 對中間數取餘
final int i = this.whichItemWorst.incrementAndGet() % half;
return tmpList.get(i).getName();
}
}
return null;
}
}
再回到MQFaultStrategy中選擇訊息佇列的地方，在開啟故障延遲機制的時候，選擇佇列後會呼叫LatencyFaultToleranceImpl的isAvailable方法來判斷Broker是否可用，而LatencyFaultToleranceImpl的isAvailable方法又是呼叫Broker對應 FaultItem的isAvailable方法來判斷的。
由上面的分析可知，isAvailable返回true表示Broker已經過了規避時間可以用於傳送訊息，返回false表示還在規避時間內，需要避免選擇此Broker，所以故障延遲機制指的是在傳送訊息時記錄每個Broker的耗時時間，如果某個Broker發生故障，但是生產者還未感知（NameServer 30s檢測一次心跳，有可能Broker已經發生故障但未到檢測時間，所以會有一定的延遲），用耗時時間做為一個故障規避時間（也可以是30000ms），此時訊息會傳送失敗，在重試或者下次選擇訊息佇列的時候，如果在規避時間內，可以在短時間內避免再次選擇到此Broker，以此達到故障規避的目的。
如果某個主題所在的所有Broker都處於不可用狀態，此時呼叫pickOneAtLeast方法儘量選擇延遲時間最短、規避時間最短（排序後的失敗條目中靠前的元素）的Broker作為此次發生訊息的Broker。
public class MQFaultStrategy {
private final LatencyFaultTolerance<String> latencyFaultTolerance = new LatencyFaultToleranceImpl();
/**
* 選擇訊息佇列
* @param tpInfo 主題路由資訊
* @param lastBrokerName 上一次使用的Broker名稱
* @return
*/
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
// 如果啟用故障延遲機制
if (this.sendLatencyFaultEnable) {
try {
// 計數器增1
int index = tpInfo.getSendWhichQueue().incrementAndGet();
// 遍歷TopicPublishInfo中儲存的訊息佇列列表
for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
// 輪詢選擇一個訊息佇列
int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
// 如果下標小於0，則使用0
if (pos < 0)
pos = 0;
// 根據下標獲取訊息佇列
MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
// 判斷訊息佇列所屬的Broker是否可用,如果可用返回當前選擇的訊息佇列
if (latencyFaultTolerance.isAvailable(mq.getBrokerName()))
return mq;
}
// 如果未獲取到可用的Broker
// 呼叫pickOneAtLeast選擇一個
final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
// 從tpInfo中獲取Broker可寫的佇列數量
int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
// 如果可寫的佇列數量大於0
if (writeQueueNums > 0) {
// 選擇一個訊息佇列
final MessageQueue mq = tpInfo.selectOneMessageQueue();
if (notBestBroker != null) {
// 設定訊息佇列所屬的Broker
mq.setBrokerName(notBestBroker);
// 設定佇列ID
mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums);
}
// 返回訊息佇列
return mq;
} else {
// 移除Broker
latencyFaultTolerance.remove(notBestBroker);
}
} catch (Exception e) {
log.error("Error occurred when selecting message queue", e);
}
// 如果故障延遲機制未選出訊息佇列，呼叫selectOneMessageQueue選擇訊息佇列
return tpInfo.selectOneMessageQueue();
}
// 根據上一次使用的BrokerName獲取訊息佇列
return tpInfo.selectOneMessageQueue(lastBrokerName);
}
}
public class LatencyFaultToleranceImpl implements LatencyFaultTolerance<String> {
private final ConcurrentHashMap<String, FaultItem> faultItemTable = new ConcurrentHashMap<String, FaultItem>(16);
@Override
public boolean isAvailable(final String name) {
final FaultItem faultItem = this.faultItemTable.get(name);
if (faultItem != null) {
// 呼叫FaultItem的isAvailable方法判斷是否可用
return faultItem.isAvailable();
}
return true;
}
}
參考
丁威、周繼鋒《RocketMQ技術內幕》
RocketMQ版本：4.9.3