RocketMq消息发送的三种方式—Default、Kernel、Selector源码级理解
2023-12-21 05:37:23
消息发送方式:Default、Kernel、Selector
三者的区别
Kernel方式发送消息是底层的方式,致力于发送一条message到指定的MessageQueue中。
Default方式发送消息是上一层的发送方式,它没有指定特定的MessageQueue,而是通过一系列逻辑之后达到对Broker和MessageQueue轮询的目的,将message发送到一个经过轮询判断之后的MessageQueue中。
Selector方式发送消息是通过选择器来选择一个特定的MessageQueue来接收发送的message,其也是一个上层发送方式,是在Kernel方式的基础上加了一层选择器逻辑。
DefaultMQProducerImpl
sendDefaultImpl()
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 maxTimeout = this.defaultMQProducer.getSendMsgTimeout() + 1000;
final long beginTimestamp = System.currentTimeMillis();
long endTimestamp = beginTimestamp;
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
if (topicPublishInfo != null && topicPublishInfo.ok()) {
// 可以看到变量 mq 是定义在循环外的,所以在第一次正常投递 Message 时,它肯定为 null。只有在第 2、3 次循环时 mq 才有值,而进行到了 2、3 次就说明首次投递失败,需要重新进行选择。
MessageQueue mq = null;
Exception exception = null;
SendResult sendResult = null;
int timesTotal = 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed();
int times = 0;
// 记录投递的BrokerName
String[] brokersSent = new String[timesTotal];
for (; times < timesTotal && (endTimestamp - beginTimestamp) < maxTimeout; times++) {
// 上一个BrokerName是从mq中取出的,一个topic下可存在多个mq,这些多个可存在于不同的broker中,因此,lastBrokerName是不同的(由于mq采用轮询方式负载均衡,导致broker也是轮询方式负载均衡,这一点详细参考selectOneMessageQueue()方法)
String lastBrokerName = null == mq ? null : mq.getBrokerName();
// 若当前message发送到mq失败,则轮询获取位于不同broker上相同topic的mq来达到对broker的轮询,以便于确保message发送的成功率。
MessageQueue tmpmq = topicPublishInfo.selectOneMessageQueue(lastBrokerName);
if (tmpmq != null) {
mq = tmpmq;
brokersSent[times] = mq.getBrokerName();
try {
// 确定mq之后,调用底层发送message的方法。
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
endTimestamp = System.currentTimeMillis();
switch (communicationMode) {
// 在异步和单工通信方式的场景下,只需要发送消息即可,不关心返回值,所以可以return null.
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) {
log.warn("sendKernelImpl exception", e);
log.warn(msg.toString());
exception = e;
endTimestamp = System.currentTimeMillis();
continue;
}
catch (MQClientException e) {
log.warn("sendKernelImpl exception", e);
log.warn(msg.toString());
exception = e;
endTimestamp = System.currentTimeMillis();
continue;
}
catch (MQBrokerException e) {
log.warn("sendKernelImpl exception", e);
log.warn(msg.toString());
exception = e;
endTimestamp = System.currentTimeMillis();
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;
default:
if (sendResult != null) {
return sendResult;
}
throw e;
}
}
catch (InterruptedException e) {
log.warn("sendKernelImpl exception", e);
log.warn(msg.toString());
throw e;
}
}
else {
break;
}
} // end of for
if (sendResult != null) {
return sendResult;
}
String info =
String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s", //
times, //
(System.currentTimeMillis() - beginTimestamp), //
msg.getTopic(),//
Arrays.toString(brokersSent));
info += FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);
throw new MQClientException(info, exception);
}
List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList();
if (null == nsList || nsList.isEmpty()) {
// 说明没有设置Name Server地址
throw new MQClientException("No name server address, please set it."
+ FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL), null);
}
throw new MQClientException("No route info of this topic, " + msg.getTopic()
+ FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO), null);
}
sendSelectImpl()
private SendResult sendSelectImpl(//
Message msg,//
MessageQueueSelector selector,//
Object arg,//
final CommunicationMode communicationMode,//
final SendCallback sendCallback, final long timeout//
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
// 有效性检查
this.makeSureStateOK();
Validators.checkMessage(msg, this.defaultMQProducer);
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
if (topicPublishInfo != null && topicPublishInfo.ok()) {
MessageQueue mq = null;
try {
// 通过选择器选择一个特定的mq
mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
}
catch (Throwable e) {
throw new MQClientException("select message queue throwed exception.", e);
}
if (mq != null) {
return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
}
else {
throw new MQClientException("select message queue return null.", null);
}
}
throw new MQClientException("No route info for this topic, " + msg.getTopic(), null);
}
sendKernelImpl()
private SendResult sendKernelImpl(final Message msg,//
final MessageQueue mq,//
final CommunicationMode communicationMode,//
final SendCallback sendCallback,//
final long timeout) throws MQClientException, RemotingException, MQBrokerException,
InterruptedException {
String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
if (null == brokerAddr) {
// TODO 此处可能对Name Server压力过大,需要调优
tryToFindTopicPublishInfo(mq.getTopic());
brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
}
SendMessageContext context = null;
if (brokerAddr != null) {
byte[] prevBody = msg.getBody();
try {
int sysFlag = 0;
if (this.tryToCompressMessage(msg)) {
sysFlag |= MessageSysFlag.CompressedFlag;
}
final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
sysFlag |= MessageSysFlag.TransactionPreparedType;
}
// 发消息之前,读写权限控制时调用 Hook
if (hasCheckForbiddenHook()) {
CheckForbiddenContext checkForbiddenContext = new CheckForbiddenContext();
checkForbiddenContext.setNameSrvAddr(this.defaultMQProducer.getNamesrvAddr());
checkForbiddenContext.setGroup(this.defaultMQProducer.getProducerGroup());
checkForbiddenContext.setCommunicationMode(communicationMode);
checkForbiddenContext.setBrokerAddr(brokerAddr);
checkForbiddenContext.setMessage(msg);
checkForbiddenContext.setMq(mq);
checkForbiddenContext.setUnitMode(this.isUnitMode());
this.executeCheckForbiddenHook(checkForbiddenContext);
}
// 执行hook
if (this.hasSendMessageHook()) {
context = new SendMessageContext();
context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
context.setCommunicationMode(communicationMode);
context.setBornHost(this.defaultMQProducer.getClientIP());
context.setBrokerAddr(brokerAddr);
context.setMessage(msg);
context.setMq(mq);
this.executeSendMessageHookBefore(context);
}
SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
requestHeader.setTopic(msg.getTopic());
requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
requestHeader.setQueueId(mq.getQueueId());
requestHeader.setSysFlag(sysFlag);
requestHeader.setBornTimestamp(System.currentTimeMillis());
requestHeader.setFlag(msg.getFlag());
requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
requestHeader.setReconsumeTimes(0);
requestHeader.setUnitMode(this.isUnitMode());
if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
if (reconsumeTimes != null) {
requestHeader.setReconsumeTimes(new Integer(reconsumeTimes));
MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
}
}
SendResult sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(//
brokerAddr,// 1
mq.getBrokerName(),// 2
msg,// 3
requestHeader,// 4
timeout,// 5
communicationMode,// 6
sendCallback// 7
);
// 执行hook
if (this.hasSendMessageHook()) {
context.setSendResult(sendResult);
this.executeSendMessageHookAfter(context);
}
return sendResult;
}
catch (RemotingException e) {
// 执行hook
if (this.hasSendMessageHook()) {
context.setException(e);
this.executeSendMessageHookAfter(context);
}
throw e;
}
catch (MQBrokerException e) {
// 执行hook
if (this.hasSendMessageHook()) {
context.setException(e);
this.executeSendMessageHookAfter(context);
}
throw e;
}
catch (InterruptedException e) {
// 执行hook
if (this.hasSendMessageHook()) {
context.setException(e);
this.executeSendMessageHookAfter(context);
}
throw e;
}
finally {
msg.setBody(prevBody);
}
}
throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}
MQ的选择方式-TopicPublishInfo
selectOneMessageQueue()
通过sendWhichQueue.getAndIncrement()获得一个随机值并保存,之后每次调用会使index值++
假设第一次随机到的是index=11,队列数为5,那么采用取余计算11%5为1,所以第一次会选择下标为1的队列,之后每次调用都会使index++,选择的队列顺序为:1,2,3,4,0,1,2,3,4,0
实现了线性轮询的负载均衡算法
它可以将流量均匀地分发给不同的 MessageQueue,而 MessageQueue 分布在不同的 Broker 上,这样也达到了对最终 Message 存储的负载均衡,避免造成数据倾斜。
/**
* 如果lastBrokerName不为null,则寻找与其不同的MessageQueue
*/
public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
if (lastBrokerName != null) {
int index = this.sendWhichQueue.getAndIncrement();
for (int i = 0; i < this.messageQueueList.size(); i++) {
int pos = Math.abs(index++) % this.messageQueueList.size();
MessageQueue mq = this.messageQueueList.get(pos);
if (!mq.getBrokerName().equals(lastBrokerName)) {
return mq;
}
}
return null;
}
else {
int index = this.sendWhichQueue.getAndIncrement();
int pos = Math.abs(index) % this.messageQueueList.size();
return this.messageQueueList.get(pos);
}
}
getAndIncrement()
此方法是一个native方法,具体逻辑是第一次随机获取一个index,之后每次获取都在此index的基础上 + 1
文章来源:https://blog.csdn.net/weixin_52457624/article/details/135114278
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符,请联系我的编程经验分享网邮箱:veading@qq.com进行投诉反馈,一经查实,立即删除!
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符,请联系我的编程经验分享网邮箱:veading@qq.com进行投诉反馈,一经查实,立即删除!