1. Preface
Apache RocketMQ, as a well-known open source messaging middleware, was born in Alibaba and donated to Apache in 2016. From RocketMQ 4.0 to the latest v4 7.1 Alibaba has won wide attention and praise both inside and outside the community.
Out of interest and work needs, I recently studied some codes of RocketMQ 4.7.1, which caused a lot of confusion and gained more inspiration.
From the perspective of the sender, this paper will analyze how RocketMQ works in transaction message sending by reading the source code of RocketMQ Producer. It should be noted that the codes posted in this article are from the RocketMQ source code of version 4.7.1. The sending discussed in this article only refers to the process of sending messages from the Producer to the Broker, and does not include the process of the Broker delivering messages to the Consumer.
2. Macro overview
RocketMQ transaction message sending process:
Figure 1
Combined with the source code, the sendMessageInTransaction method of the transaction message TransactionMQProducer of RocketMQ actually calls the sendMessageInTransaction method of defaultmqproduceriml. We enter the sendMessageInTransaction method, and the sending process of the whole transaction message is clearly visible:
First, check before sending and fill in the necessary parameters, including setting the prepare transaction message.
Source code List-1
public TransactionSendResult sendMessageInTransaction(final Message msg,
final LocalTransactionExecuter localTransactionExecuter, final Object arg)
throws MQClientException {
TransactionListener transactionListener = getCheckListener();
if (null == localTransactionExecuter && null == transactionListener) {
throw new MQClientException("tranExecutor is null", null);
}
// ignore DelayTimeLevel parameter
if (msg.getDelayTimeLevel() != 0) {
MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
}
Validators.checkMessage(msg, this.defaultMQProducer);
SendResult sendResult = null;
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());Enter the sending process:
Source code list-2
try {
sendResult = this.send(msg);
} catch (Exception e) {
throw new MQClientException("send message Exception", e);
}Decide whether to execute the local transaction according to the processing results returned by the broker. If the semi message is sent successfully, start the local transaction execution:
Source code list-3
LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
Throwable localException = null;
switch (sendResult.getSendStatus()) {
case SEND_OK: {
try {
if (sendResult.getTransactionId() != null) {
msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
}
String transactionId = msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
if (null != transactionId && !"".equals(transactionId)) {
msg.setTransactionId(transactionId);
}
if (null != localTransactionExecuter) {
localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
} else if (transactionListener != null) {
log.debug("Used new transaction API");
localTransactionState = transactionListener.executeLocalTransaction(msg, arg);
}
if (null == localTransactionState) {
localTransactionState = LocalTransactionState.UNKNOW;
}
if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) {
log.info("executeLocalTransactionBranch return {}", localTransactionState);
log.info(msg.toString());
}
} catch (Throwable e) {
log.info("executeLocalTransactionBranch exception", e);
log.info(msg.toString());
localException = e;
}
}
break;
case FLUSH_DISK_TIMEOUT:
case FLUSH_SLAVE_TIMEOUT:
case SLAVE_NOT_AVAILABLE: // When the standby broker status is unavailable, the half message will be rolled back and the local transaction will not be executed
localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
break;
default:
break;
}After the local transaction is completed, two-stage processing is carried out according to the local transaction status:
Source code list-4
try {
this.endTransaction(sendResult, localTransactionState, localException);
} catch (Exception e) {
log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e);
}
// Assembly sending results
// ...
return transactionSendResult;
}Next, let's go deep into the code analysis of each stage.
3. Dig deep inside
3.1 phase I transmission
Focus on the analysis of send method. After entering the send method, we find that the SYNC synchronization mode is used in the transaction message phase of RocketMQ:
Source code list-5
public SendResult send(Message msg,
long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
}This is easy to understand. After all, the transaction message needs to decide whether to execute the local transaction according to the result of one-stage sending, so the ack waiting for the broker must be blocked.
We enter defaultmqproducerimpl Look at the implementation of sendDefaultImpl method in Java. Read the code of this method to try to understand the behavior of producer in the sending process of transaction messages. It is worth noting that this method is not customized for transaction messages or even SYNC synchronization mode. Therefore, after reading this code, you can basically have a more comprehensive understanding of the message sending mechanism of RocketMQ.
The logic of this code is very smooth. I can't bear to slice it. In order to save space, the more complicated but less informative part of the code is replaced by comments to preserve the integrity of the process as much as possible. The parts that I think are more important or easy to be ignored are marked with notes, and some details are explained in detail later.
Source code list-6
private SendResult sendDefaultImpl(
Message msg,
final CommunicationMode communicationMode,
final SendCallback sendCallback,
final long timeout
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
this.makeSureStateOK();
// 1, Message validity verification. See below
Validators.checkMessage(msg, this.defaultMQProducer);
final long invokeID = random.nextLong();
long beginTimestampFirst = System.currentTimeMillis();
long beginTimestampPrev = beginTimestampFirst;
long endTimestamp = beginTimestampFirst;
// Get the sending route information of the current topic, mainly from the broker. If it is not found, get it from namesrv
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
if (topicPublishInfo != null && topicPublishInfo.ok()) {
boolean callTimeout = false;
MessageQueue mq = null;
Exception exception = null;
SendResult sendResult = null;
// 2, Send retry mechanism. See below
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
int times = 0;
String[] brokersSent = new String[timesTotal];
for (; times < timesTotal; times++) {
// The first sending is mq == null, and then there is broker information
String lastBrokerName = null == mq ? null : mq.getBrokerName();
// 3, How to select a queue when rocketmq sends a message—— broker exception avoidance mechanism
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;
}
// Send core code
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
endTimestamp = System.currentTimeMillis();
// The circumvention mechanism when rocketmq selects a broker. It takes effect only when sendLatencyFaultEnable == true
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
switch (communicationMode) {
// 4, Three communication modes of RocketMQ. See below
case ASYNC: // Asynchronous mode
return null;
case ONEWAY: // Unidirectional mode
return null;
case SYNC: // Synchronous mode
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
continue;
}
}
return sendResult;
default:
break;
}
} catch (RemotingException e) {
// ...
// automatic retry
} catch (MQClientException e) {
// ...
// automatic retry
} catch (MQBrokerException e) {
// ...
// Return code only = = not_ IN_ CURRENT_ Automatically retry when unit = = 205
// Do not try again in other cases, throw exceptions
} catch (InterruptedException e) {
// ...
// No retry, throw exception
}
} else {
break;
}
}
if (sendResult != null) {
return sendResult;
}
// The info information returned by the assembly is finally thrown with MQClientException
// ... ...
// Throw remotingtoomuchrequexexception in timeout scenario
if (callTimeout) {
throw new RemotingTooMuchRequestException("sendDefaultImpl call timeout");
}
// Fill in MQClientException exception information
// ...
}
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);
}3.1.1 message validity verification
Source code list-7
Validators.checkMessage(msg, this.defaultMQProducer);
In this method, the validity of the message is verified, including the verification of topic and message body. The naming of topic must comply with the specification, and the use of built-in system message topic should be avoided. Message body length > 0 & & message body length < = 102410244 = 4m.
Source code list-8
public static void checkMessage(Message msg, DefaultMQProducer defaultMQProducer)
throws MQClientException {
if (null == msg) {
throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message is null");
}
// topic
Validators.checkTopic(msg.getTopic());
Validators.isNotAllowedSendTopic(msg.getTopic());
// body
if (null == msg.getBody()) {
throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body is null");
}
if (0 == msg.getBody().length) {
throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL, "the message body length is zero");
}
if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) {
throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL,
"the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize());
}
}3.1.2 send retry mechanism
Producer will automatically retry when the message is not sent successfully. The maximum sending times = retryTimesWhenSendFailed + 1 = 3 times.
It is worth noting that not all exceptions will be retried. The information extracted from the above source code tells us that it will be retried automatically in the following three cases:
1) One of RemotingException and MQClientException occurs.
2) An MQBrokerException exception occurred and the ResponseCode is not_ IN_ CURRENT_ When unit = 205.
3) In SYNC mode, no exception occurs and the sending result status is not SEND_OK.
Before sending a message each time, it will check whether the previous two steps have taken too long (the timeout duration is 3000ms by default). If so, it will not continue to send, and the timeout will be returned directly without retry. Here are two questions:
1) The internal automatic retry of producer is imperceptible to the business application, and the sending time seen by the application includes the time of all retries;
2) Once timeout occurs, it means that this message sending has ended in failure. The reason is timeout. This information will finally be thrown in the form of remotingtoomuchrequexexception.
It should be pointed out here that the official document of rocketmq indicates that the transmission timeout is 10s, i.e. 10000ms. Many people on the Internet also think that the timeout of rocketmq is 10s. However, 3000ms is clearly written in the code. Finally, I confirmed after debug ging that the default timeout is 3000ms. It is also recommended that the rocketmq team confirm the document. If there is any error, it is better to correct it as soon as possible.
Figure 2
3.1.3 exception avoidance mechanism of broker
Source code list-8
MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
This line of code is the process of selecting queue before sending.
This involves a core mechanism of RocketMQ message sending high availability, latencyFaultTolerance. This mechanism is a part of Producer's load balancing and is controlled by the value of sendLatencyFaultEnable. The default is false. The broker failure delay mechanism is not started. When the value is true, the broker failure delay mechanism is enabled and can be actively opened by Producer.
When selecting a queue, turn on the exception avoidance mechanism. According to the working state of the broker, avoid selecting the broker agent with poor current state. The unhealthy broker will be avoided for a period of time. When the exception avoidance mechanism is not turned on, select the next queue in order, but in the retry scenario, try to select a queue different from the last sent broker. Every time a message is sent, the status information of the broker will be maintained through the updatefaultetem method.
Source code list-9
public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
if (this.sendLatencyFaultEnable) {
// Calculate how long the delay is. isolation indicates whether the broker needs to be isolated. If so, find the first delay value smaller than 30s from 30s, and then judge the avoidance period according to the subscript. If 30s, it is 10min;
// Otherwise, the evasion time is determined according to the last transmission time;
long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
}
} Go deep into the selectOneMessageQueue method to find out:
Source code list-10
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
if (this.sendLatencyFaultEnable) {
// Enable exception avoidance
try {
int index = tpInfo.getSendWhichQueue().getAndIncrement();
for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
if (pos < 0)
pos = 0;
// Take down a message queue in sequence as the sent queue
MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
// The broker of the current queue is available and is the same as the broker of the previous queue,
// Or this queue is used for the first time
if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
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();
}
// If exception avoidance is not enabled, Queue is selected randomly
return tpInfo.selectOneMessageQueue(lastBrokerName);
}3.1.4 three communication modes of rocketmq
Source code list-11
public enum CommunicationMode {
SYNC,
ASYNC,
ONEWAY,
}The above three modes refer to the stage when the message arrives at the broker from the sender, and do not include the process of the broker delivering the message to the subscriber.
Differences in transmission modes of the three modes:
- One way mode: ONEWAY. The message sender just sends the message and doesn't care about the result of the broker processing. In this mode, due to the small processing flow, the sending time is very small and the throughput is large, but it can not guarantee the reliability of the message. It is often used in the scenario of large traffic but unimportant message, such as heartbeat sending.
- Asynchronous mode: ASYNC. After the message sender sends the message to the broker, it does not need to wait for the broker to process. What it gets is a null return value. Instead, an asynchronous thread processes the message. After processing, it tells the sender the sending result in the form of callback. If there is an exception during asynchronous processing, the sender will retry internally before returning the failure result of the sender (3 times by default, which is not perceived by the sender). In this mode, the sender's waiting time is small, the throughput is large, and the message is reliable. It is used in large traffic but important message scenarios.
- Synchronization mode: SYNC. The message sender needs to wait for the broker to complete processing and clearly return success or failure. Before the message sender gets the result of message sending failure, it will also experience internal retry (3 times by default, which is not perceived by the sender). In this mode, the sender will block and wait for the message processing result. The waiting time is long and the message is reliable. It is used in the message scenario with small traffic but important. It should be emphasized that the processing of one-stage and half transaction messages is synchronous mode.
Specific implementation differences can also be seen in the sendKernelImpl method. ONEWAY mode is the simplest and does not do any processing. In the sendMessage method parameters responsible for sending, compared with the synchronous mode, the asynchronous mode has more callback methods, topicPublishInfo containing topic sending route meta information, instance containing sending broker information, producer containing sending queue information and retry times. In addition, in the asynchronous mode, the compressed messages will be copied first.
Source code list-12
switch (communicationMode) {
case ASYNC:
Message tmpMessage = msg;
boolean messageCloned = false;
if (msgBodyCompressed) {
//If msg body was compressed, msgbody should be reset using prevBody.
//Clone new message using commpressed message body and recover origin massage.
//Fix bug:https://github.com/apache/rocketmq-externals/issues/66
tmpMessage = MessageAccessor.cloneMessage(msg);
messageCloned = true;
msg.setBody(prevBody);
}
if (topicWithNamespace) {
if (!messageCloned) {
tmpMessage = MessageAccessor.cloneMessage(msg);
messageCloned = true;
}
msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQProducer.getNamespace()));
}
long costTimeAsync = System.currentTimeMillis() - beginStartTime;
if (timeout < costTimeAsync) {
throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
}
sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
mq.getBrokerName(),
tmpMessage,
requestHeader,
timeout - costTimeAsync,
communicationMode,
sendCallback,
topicPublishInfo,
this.mQClientFactory,
this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(),
context,
this);
break;
case ONEWAY:
case SYNC:
long costTimeSync = System.currentTimeMillis() - beginStartTime;
if (timeout < costTimeSync) {
throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");
}
sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
mq.getBrokerName(),
msg,
requestHeader,
timeout - costTimeSync,
communicationMode,
context,
this);
break;
default:
assert false;
break;
} There is such a picture in the official document, which clearly describes the detailed process of asynchronous communication:
Figure 3
3.2 phase II transmission
Listing-3 of the source code reflects the execution of local transactions. localTransactionState associates the execution results of local transactions with the sending of transaction messages in the second phase.
It is worth noting that if the sending result of the first stage is SLAVE_NOT_AVAILABLE, that is, when the standby broker is unavailable, the localTransactionState will also be set to Rollback, and the local transaction will not be executed at this time. After that, the endTransaction method is responsible for the two-stage submission. See the source code listing - 4. Specific to the implementation of endTransaction:
Source code list-13
public void endTransaction(
final SendResult sendResult,
final LocalTransactionState localTransactionState,
final Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {
final MessageId id;
if (sendResult.getOffsetMsgId() != null) {
id = MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());
} else {
id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
}
String transactionId = sendResult.getTransactionId();
final String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
requestHeader.setTransactionId(transactionId);
requestHeader.setCommitLogOffset(id.getOffset());
switch (localTransactionState) {
case COMMIT_MESSAGE:
requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);
break;
case ROLLBACK_MESSAGE:
requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);
break;
case UNKNOW:
requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);
break;
default:
break;
}
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
requestHeader.setMsgId(sendResult.getMsgId());
String remark = localException != null ? ("executeLocalTransactionBranch exception: " + localException.toString()) : null;
// Use oneway to send two-stage messages
this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark,
this.defaultMQProducer.getSendMsgTimeout());
}In two-stage sending, I understand that the reason why I use oneway to send is because the transaction message has a special reliable mechanism - backcheck.
3.3 message reply
After a specific time, the Broker finds that it still does not get the exact information about whether the second phase of the transaction message should be committed or rolled back. The Broker does not know what happened to the producer (the producer may hang up, or the producer may send a commit, but the network jitter is lost, or...), So he took the initiative to initiate a backcheck.
The check back mechanism of transaction messages is more embodied on the broker side. The broker of RocketMQ isolates the transaction messages in different sending stages with three different topic s: Half message, Op message and real message, so that the Consumer can only see the messages that need to be delivered after the final confirmation commit. The detailed implementation logic will not be repeated in this article for the time being. A separate article can be opened later to interpret it from the perspective of broker.
Back to the Producer's perspective, when receiving the Broker's backcheck request, the Producer will check the local transaction status according to the message and decide to commit or rollback according to the result, which requires the Producer to specify the backcheck implementation in case of need.
Of course, under normal circumstances, it is not recommended to actively send the unknown state. This state will undoubtedly bring additional backcheck overhead to the broker. It is a reasonable choice to start the backcheck mechanism only in case of unpredictable exceptions.
In addition, the transaction backcheck in version 4.7.1 is not unlimited, but up to 15 times:
Source code list-14
/** * The maximum number of times the message was checked, if exceed this value, this message will be discarded. */ @ImportantField private int transactionCheckMax = 15;
appendix
The default parameters of Producer given by the official are as follows (the parameter of timeout duration has also been mentioned earlier, and the result of debug is 3000ms by default, not 10000ms):
Figure 4
As an excellent open source message middleware, RocketMQ has been redeveloped by many developers based on it. For example, SOFAStack MQ message queue, a commercial product of ant group, is a financial level message middleware redeveloped based on RocketMQ kernel. It has done a lot of excellent work in message control, transparent operation and maintenance, etc.
I hope RocketMQ will continue to grow and burst into stronger vitality under the joint creation and joint construction of the majority of developers in the community.
We are Alibaba cloud intelligent global technology service SRE team. We are committed to becoming a technology-based, service-oriented and high availability engineer team; Provide professional and systematic SRE services to help customers better use the cloud, build more stable and reliable business systems based on the cloud, and improve business stability. We look forward to sharing more technologies that can help enterprise customers get on the cloud, make good use of the cloud, and make their business on the cloud run more stable and reliable. You can scan the QR code below with nails, join the nail circle of Alibaba cloud SRE Institute of technology, and communicate with more people on the cloud about the cloud platform.
Original link: https://developer.aliyun.com/article/783843?
Copyright notice: the content of this article is spontaneously contributed by Alibaba cloud real name registered users, and the copyright belongs to the original author. Alibaba cloud developer community does not own its copyright or bear corresponding legal liabilities. Please refer to Alibaba cloud developer community user service agreement and Alibaba cloud developer community intellectual property protection guidelines for specific rules. If you find any content suspected of plagiarism in the community, fill in the infringement complaint form to report. Once verified, the community will immediately delete the content suspected of infringement.