A distributed transaction means that the participants of the transaction, the server supporting the transaction, the resource server and the transaction manager are located on different nodes of different distributed systems. For example, in a large-scale e-commerce system, the ordering interface usually deducts inventory, subtracts discounts and generates order IDS. However, the order service is different from the inventory, discounts and order IDS. The success of the ordering interface depends not only on the local DB operation, but also on the results of the third-party system. A distributed transaction guarantees that all of these operations will either succeed or fail. Essentially, distributed transactions are about ensuring data consistency across databases.

At present, seATA, LCN and other solutions to solve distributed things.

RocketMQ distributed transaction implementation

RocketMQ provides the function of transaction message, and adopts the distributed transaction function of 2PC(two-segment protocol)+ compensation mechanism (transaction backcheck). The final agreement of distributed transaction can be achieved through the message queue of RocketMQ version transaction message.

First, we need to know what is semi-thing message and message back check:

  • Semi-transactional message:

    For a message that is temporarily undeliverable, the sender has successfully sent the message to the RocketMQ server of the message queue, but the server has not received the second acknowledgement of the message from the producer. At this time, the message is marked as “temporarily undeliverable”. A message in this state is a semi-transactional message.

  • Message back check:

    The second acknowledgement of a transaction message is lost due to network disconnection or producer application restart. When the RocketMQ server detects that a message has been “semi-transactional” for a long time through scanning, it needs to proactively ask the producer for the final status of the message (Commit or Rollback). This interrogation process is called message lookup.

[Interactive process]

The procedure for sending a transaction message is as follows:

  1. The sender sends the semi-transactional message to the RocketMQ version of the message queue server.

  2. Message queue The RocketMQ version server returns an Ack to the sender after successfully persisting the message. Confirm that the message has been sent successfully, and the message is semi-transactional.

  3. The sender begins to execute the local transaction logic.

  4. The sender submits a second acknowledgement (Commit or Rollback) to the server based on the result of the local transaction execution. When the server receives the Commit status, it marks the semi-transactional message as deliverable, and the subscriber eventually receives the message. When the server receives the Rollback state, it deletes the semi-transactional message and the subscriber will not accept the message.

The procedure for retrospecting transaction messages is as follows:

  1. In the special case of network disconnection or application restart, if the secondary acknowledgement submitted in step 4 does not reach the server, the server will check the message after a fixed period of time.

  2. After the sender receives the message and checks back, it needs to check the final result of the local transaction execution of the corresponding message.

  3. The sender submits a second acknowledgement based on the final status of the local transaction, and the server still performs operations on the semi-transaction message according to Step 4.

In general, RocketMQ transaction messages are divided into two main lines:

  • Sending process: Send half message, execute local transaction, and send transaction execution result

  • Periodic task check process: The MQ periodic task scans half messages, checks local transactions, and sends transaction execution results

The source code related

Prepare of Producer sending transaction semi-message

The core class for sending transaction messages in local applications is TransactionMQProducer. This class replicates most of the logic associated with sending messages by inheriting DefaultMQProducer. This class has a very small code size of about 100 lines. Here is the sendMessageTransaction method for this class

The transactionListener class provides two methods:

  • executeLocalTransaction

    Perform local transactions

  • checkLocalTransaction

    Check back local transactions

Then see DefaultMQProducer. SendMessageInTransaction () method:

This method basically does the following

  • Messages are tagged with a transaction message-related tag that the broker uses to distinguish between ordinary and transactional messages

  • Sending a half Message

  • If the send is successful, the local transaction is executed by transactionListener

  • Execute the endTransaction method to tell the broker to perform COMMIT /rollback.

Perform local transactions

Producer half a transaction message is sent after the success, will call transactionListener. ExecuteLocalTransaction methods perform local affairs. The local transaction is executed only after the half-message is successfully sent. If the half-message fails, the rollback is set.

End transaction (COMMIT /rollback)

After the local transaction executes, the this.endTransaction() method is called to commit or roll back the transaction, depending on the state of the local transaction execution. If the half-message fails to be sent or the local transaction fails to be executed, it tells the server that the half-message is deleted. If the half-message is sent successfully and the local transaction succeeds, it tells the server that the half-message takes effect

The Broker side processes the transactional messages

Broker side by SendMessageProcessor. The processRequest () method takes processing Producer sends a message Finally will be called to SendMessageProcessor. The sendMessage (), message type, Do message storage.

Stored semi-message

Code prepareMessage (msgInner) :

In this step, the original topic name of the message is backed up with the original queue ID, then the message label of the transaction message is unlabeled and the topic of the message is reset to RMQ_SYS_TRANS_HALF_TOPIC with the queue ID fixed to 0. Distinguish it from other normal messages, and then complete message persistence. At this point, the Broker has initially processed the transaction semi-message sent by the Producer.

Semi-message transaction lookup

A two-stage protocol sends and commits a rollback message. When the status of the local transaction message is unknown, the transaction is terminated and no operation is performed. The transaction status of the sender is ROLLBACK or COMMIT through periodic backcheck. Through TransactionalMessageCheckService thread regularly to detect RMQ_SYS_TRANS_HALF_TOPIC topics in the news, back to check the message state of affairs.

  • RMQ_SYS_TRANS_HALF_TOPIC

    The topic of the PREPARE message to which the transaction message first enters.

  • RMQ_SYS_TRANS_OP_HALF_TOPIC

    When the message server receives a commit or rollback request for a transaction message, the message is stored under the topic.

The Broker handle END_TRANSACTION

When a transaction is committed or rolled back by a Producer or a periodic task, how does the Broker handle the commit and rollback commands? Its core implementation is as follows:

  • The message was found according to commitlogOffset

  • In the case of a commit action, the subject and queue of the original message are restored, re-stored in the commitlog file and then transferred to the message consumption queue for consumption by the consumer, and then the original preprocessed message is stored in a new topic, RMQ_SYS_TRANS_OP_HALF_TOPIC, indicating that the message has been processed

  • If the message is rolled back, the original preprocessed message is directly stored into a new topic, RMQ_SYS_TRANS_OP_HALF_TOPIC, indicating that the message has been processed.

Overall implementation process

What if the consumer fails?

If any message consumption fails, the failed message is sent back to the broker, that is, re-written to the commitLog file, and the consumer re-consumes. If it comes back, the disconnect between the consumer and the broker network, consumer will call submitConsumeRequestLater () method, back on consumer spending, if consumer still fails, will continue to try again until you reach the default 16 times, You can use the MSG. GetReconsumeTimes () method to obtain the current number of retries. If the number of retries is enough and the consumption is still not successful, you must manually intervene in the form of work order and log to roll back the producer transaction.

The Producer failed to send a semi-message

The order system of Producer failed to send prepare due to network or MQ faults. Procedure At this time, the order system can perform rollback operations, such as “order closed”, and refund to the user through the reverse process.

The semi-message was sent successfully, but the local transaction failed

If the semi-message sent by the order system succeeds, but the execution of the local transaction fails, such as updating the order status to “completed”. In this case, when the local transaction fails, rollback is returned to MQ, which deletes the previously sent half-message. I wouldn’t have called the coupon system.

Semi-message sent successfully, no response from MQ was received

Suppose the order system sends a semi-message successfully and does not receive a response from MQ. At this time, it may be due to network problems or other abnormal errors. The order system mistakenly thought that sending MQ semi-message failed and executed the reverse rollback process. But at this time actually MQ has saved the semi-message successfully, then how to deal with this message? This time background of MQ message back to check timing task TransactionalMessageCheckService will scan every 1 minutes one and a half times the message queue, to determine whether you need the message back to check, and then back to check the order system local transactions, the MQ will find order has become a “closed”, At this point, a ROLLBACK request is sent to MQ to remove the previous half-message.

If commit/rollback fails

This is by task TransactionalMessageCheckService regularly, it will find the news over a certain period of time has not yet two phase processing, will be back to check local transactions.

summary

Message queues RocketMQ distributed transaction messages not only decouples applications, but also ensures the ultimate consistency of data. At the same time, traditional large transactions can be broken down into small transactions, which not only improves efficiency, but also prevents the whole system from being rolled back due to the unavailability of a certain associated application, thus maximizing the availability of the core system. In extreme cases, if an associated application fails to be processed successfully, only compensation or data correction is required for the current application instead of rolling back the entire business.

The RocketMQ transactional message link embodies the failure-oriented design idea and the strictness of the transactional system. When the messages in the second stage are not delivered, the broker will actively request the producer to do the check, and the producer will return the transaction state again after the check. While there are many ways to achieve ultimate consistency, transactional messaging is one of the more elegant ways to do it.