RocketMQ中如何實(shí)現(xiàn)producer消息發(fā)送
這篇文章主要介紹了RocketMQ中如何實(shí)現(xiàn)producer消息發(fā)送，具有一定借鑒價值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。
在阿合奇等地區(qū)，都構(gòu)建了全面的區(qū)域性戰(zhàn)略布局，加強(qiáng)發(fā)展的系統(tǒng)性、市場前瞻性、產(chǎn)品創(chuàng)新能力，以專注、極致的服務(wù)理念，為客戶提供成都網(wǎng)站建設(shè)、成都做網(wǎng)站 網(wǎng)站設(shè)計(jì)制作定制網(wǎng)站建設(shè),公司網(wǎng)站建設(shè),企業(yè)網(wǎng)站建設(shè),成都品牌網(wǎng)站建設(shè),全網(wǎng)整合營銷推廣,成都外貿(mào)網(wǎng)站建設(shè)公司,阿合奇網(wǎng)站建設(shè)費(fèi)用合理。
RocketMQ中一個topic可以分布在多個broker上，每個broker上又可以包含多個message queue。每個message具體發(fā)送到哪個broker的哪個message queue中，這個決策過程是在client端完成的。client會從nameserver訂閱topic的路由信息，按照一定的負(fù)載均衡算法為每個message選擇出一個queue，并完成投遞。producer的消息投遞過程大體如下圖：
1. 出錯重試
producer的投遞出錯重試分為兩種情況，同步出錯重試（sync）以及異步出錯重試（async）。同步發(fā)送出錯重試是在DefaultMQProducerImpl.sendDefaultImpl 中完成：
// SYNC模式則開啟重試，ASYNC模式重試在MQClientAPIImpl中實(shí)現(xiàn)
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
int times = 0;
String[] brokersSent = new String[timesTotal];
for (; times < timesTotal; times++) { // 循環(huán)重試最多timesTotal次
String lastBrokerName = null == mq ? null : mq.getBrokerName();
MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
if (mqSelected != null) {
mq = mqSelected;
brokersSent[times] = mq.getBrokerName();
try {
// ....
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);
// ...
exception = e;
continue; // 出錯，進(jìn)入下一次重試
} catch (MQClientException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
// ...
exception = e;
continue; // 出錯，進(jìn)入下一次重試
} catch (MQBrokerException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
// ...
exception = e;
switch (e.getResponseCode()) {
case ResponseCode.TOPIC_NOT_EXIST:
case ResponseCode.SERVICE_NOT_AVAILABLE:
case ResponseCode.SYSTEM_ERROR:
case ResponseCode.NO_PERMISSION:
case ResponseCode.NO_BUYER_ID:
case ResponseCode.NOT_IN_CURRENT_UNIT:
continue; // 出錯，進(jìn)入下一次重試
default:
if (sendResult != null) { // 放棄重試
return sendResult;
}
throw e;
}
} catch (InterruptedException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
// ...
throw e; // 放棄重試
}
} else {
break;
}
}異步發(fā)送的出錯重試則是在更底層的MQClientAPIImpl.sendMessageAsync里實(shí)現(xiàn)：
// 調(diào)用invokeAsync發(fā)送消息
this.remotingClient.invokeAsync(addr, request, timeoutMillis, new InvokeCallback() {
@Override
public void operationComplete(ResponseFuture responseFuture) {
long cost = System.currentTimeMillis() - beginStartTime;
RemotingCommand response = responseFuture.getResponseCommand();
// ...
if (response != null) {
try {
SendResult sendResult = MQClientAPIImpl.this.processSendResponse(brokerName, msg, response);
assert sendResult != null;
if (context != null) {
context.setSendResult(sendResult);
context.getProducer().executeSendMessageHookAfter(context);
}
try {
sendCallback.onSuccess(sendResult);
} catch (Throwable e) {
}
producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), false);
} catch (Exception e) {
producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true);
// 發(fā)送成功，processSendReponse異常重試
onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
retryTimesWhenSendFailed, times, e, context, false, producer);
}
} else {
producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true);
if (!responseFuture.isSendRequestOK()) {
MQClientException ex = new MQClientException("send request failed", responseFuture.getCause());
// 消息發(fā)送失敗，重試
onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
retryTimesWhenSendFailed, times, ex, context, true, producer);
} else if (responseFuture.isTimeout()) {
MQClientException ex = new MQClientException("wait response timeout " + responseFuture.getTimeoutMillis() + "ms",
responseFuture.getCause());
// 發(fā)送超時，重試
onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
retryTimesWhenSendFailed, times, ex, context, true, producer);
} else {
MQClientException ex = new MQClientException("unknow reseaon", responseFuture.getCause());
// 其他錯誤，重試
onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
retryTimesWhenSendFailed, times, ex, context, true, producer);
}
}
}
});在異常處理的onExceptionImpl方法中會再次觸發(fā)sendMessageAsync方法：
private void onExceptionImpl(final String brokerName,
final Message msg,
final long timeoutMillis,
final RemotingCommand request,
final SendCallback sendCallback,
final TopicPublishInfo topicPublishInfo,
final MQClientInstance instance,
final int timesTotal,
final AtomicInteger curTimes,
final Exception e,
final SendMessageContext context,
final boolean needRetry,
final DefaultMQProducerImpl producer
) {
int tmp = curTimes.incrementAndGet(); // 將已重試次數(shù)+1
if (needRetry && tmp <= timesTotal) { // 重試
String retryBrokerName = brokerName;//by default, it will send to the same broker
if (topicPublishInfo != null) { //select one message queue accordingly, in order to determine which broker to send
MessageQueue mqChosen = producer.selectOneMessageQueue(topicPublishInfo, brokerName);
retryBrokerName = mqChosen.getBrokerName();
}
String addr = instance.findBrokerAddressInPublish(retryBrokerName);
log.info("async send msg by retry {} times. topic={}, brokerAddr={}, brokerName={}", tmp, msg.getTopic(), addr,
retryBrokerName);
try {
request.setOpaque(RemotingCommand.createNewRequestId());
sendMessageAsync(addr, retryBrokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance,
timesTotal, curTimes, context, producer);
} catch (InterruptedException e1) {
onExceptionImpl(retryBrokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1,
context, false, producer);
} catch (RemotingConnectException e1) {
producer.updateFaultItem(brokerName, 3000, true);
onExceptionImpl(retryBrokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1,
context, true, producer);
} catch (RemotingTooMuchRequestException e1) {
onExceptionImpl(retryBrokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1,
context, false, producer);
} catch (RemotingException e1) {
producer.updateFaultItem(brokerName, 3000, true);
onExceptionImpl(retryBrokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1,
context, true, producer);
}
} else {
// ...
}
}2. 負(fù)載均衡
消息的負(fù)載均衡是通過MQFaultStrategy.selectOneMessageQueue方法來實(shí)現(xiàn)的：
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
if (this.sendLatencyFaultEnable) {
try {
int index = tpInfo.getSendWhichQueue().getAndIncrement();
for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
// 輪詢的方式選擇下一個queue
int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
if (pos < 0)
pos = 0;
MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
// 根據(jù)latency統(tǒng)計(jì)數(shù)據(jù)判斷是否可用
if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
// 優(yōu)先選擇同一個broker上的queue
if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
return mq;
}
}
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().getAndIncrement() % writeQueueNums);
}
return mq;
} else {
latencyFaultTolerance.remove(notBestBroker);
}
} catch (Exception e) {
log.error("Error occurred when selecting message queue", e);
}
return tpInfo.selectOneMessageQueue();
}
return tpInfo.selectOneMessageQueue(lastBrokerName);
}感謝你能夠認(rèn)真閱讀完這篇文章，希望小編分享的“RocketMQ中如何實(shí)現(xiàn)producer消息發(fā)送”這篇文章對大家有幫助，同時也希望大家多多支持創(chuàng)新互聯(lián)，關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，更多相關(guān)知識等著你來學(xué)習(xí)!
標(biāo)題名稱：RocketMQ中如何實(shí)現(xiàn)producer消息發(fā)送
標(biāo)題鏈接：http://aaarwkj.com/article20/pchjco.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供關(guān)鍵詞優(yōu)化、網(wǎng)站導(dǎo)航、標(biāo)簽優(yōu)化、用戶體驗(yàn)、全網(wǎng)營銷推廣、面包屑導(dǎo)航
聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)
- 自適應(yīng)網(wǎng)站建設(shè)有什么優(yōu)勢 2021-08-16
- 創(chuàng)新互聯(lián)介紹自適應(yīng)網(wǎng)站建設(shè)的特點(diǎn) 2016-12-15
- 普陀營銷型網(wǎng)站建設(shè)或是自適應(yīng)網(wǎng)站建設(shè)的區(qū)別 2020-11-19
- 自適應(yīng)網(wǎng)站開發(fā)seo規(guī)范和影響 2023-04-02
- 建設(shè)營銷型手機(jī)網(wǎng)站應(yīng)注意哪些要素? 2022-05-22
- 淺談自適應(yīng)網(wǎng)站建設(shè) 2021-05-21
- 企業(yè)建站是否應(yīng)該選擇自適應(yīng)網(wǎng)站設(shè)計(jì)？ 2022-05-10
- 自適應(yīng)網(wǎng)站建設(shè)-自適應(yīng)網(wǎng)站的四大注意事項(xiàng) 2016-11-13
- 自適應(yīng)網(wǎng)站建設(shè)應(yīng)該注意哪些方面問題呢? 2021-05-07
- 你知道自適應(yīng)網(wǎng)站與響應(yīng)式網(wǎng)站有什么不一樣嗎 2016-11-07
- 自適應(yīng)網(wǎng)站建設(shè)的發(fā)展趨勢 2016-10-05
- 自適應(yīng)網(wǎng)站建設(shè)時需要了解哪些問題？ 2022-10-06