Shulou(Shulou.com)06/01 Report--

This article shows you how to use RocketMQ transaction messages to solve distributed transactions, the content is concise and easy to understand, can definitely brighten your eyes, through the detailed introduction of this article, I hope you can get something.

A preliminary understanding of the core modules of RocketMQ

Rocketmq module

Rocketmq-broker: accepts the message from the producer and stores it (by calling rocketmq-store), where the consumer gets the message.

Rocketmq-client: provides the client API that sends and receives messages.

Rocketmq-namesrv:NameServer, similar to Zookeeper, holds the run-time meta-information of the message's TopicName, queue, and so on. (it smells like NameNode)

Rocketmq-common: general classes, methods, data structures, etc.

Rocketmq-remoting: client/server + fastjson serialization + Custom binary Protocol based on Netty4

Rocketmq-store: message, index storage, etc.

Rocketmq-filtersrv: message filter Server. Note that to implement this filter, you need to upload the code to MQ! [generally speaking, we can use Tag to meet most of the filtering requirements. If we have more flexible and complex filtering requirements, we can consider the filtersrv component]

Rocketmq-tools: command line tool

Distributed message queuing RocketMQ- transaction messages-solving distributed transactions

When it comes to distributed transactions, we will talk about the classic "account transfer" problem: two accounts, distributed in two different DB, or two different subsystems, A to deduct money, B to add money, how to ensure atomicity?

The general idea is to achieve "ultimate consistency" through message middleware: system A deducts money, then sends a message to middleware, system B receives the message and adds money.

But there is a question: an is to update DB first and then send a message? Or would you like to send a message first and then update DB?

Suppose the first update DB succeeds, the sending network fails, and the retransmission fails, what should I do?

Suppose the first message is sent successfully and the update DB fails. What if the message has been sent out and cannot be withdrawn?

So here's a conclusion: as long as the two operations of sending messages and update DB are not atomic, no matter who comes first, there is a problem.

Then how to solve this problem?

Wrong scenario 0

Some people may have thought that I can put the network call "send message" in the same transaction as update DB, and if the message fails, update DB rolls back automatically. Doesn't this guarantee the atomicity of the two operations?

This proposal seems to be correct, but in fact it is wrong for two reasons:

(1) the second general question of the network: does the sender not know that the message middleware really did not receive the message when it failed to send the message? Or has the message been received, but failed to return response?

If the message has been received and the sender thinks that it has not been received, the rollback operation of update db is performed. It will cause the money of An account not to be deducted, but the money of B account to be increased.

(2) putting network calls in DB transactions may lead to long DB transactions due to network delay. In serious cases, it will block the entire DB. It's a big risk.

Based on the above analysis, we know that this scheme is actually wrong! Option 1-the business side implements it by itself.

Suppose the message middleware does not provide "transaction message" functionality, such as you are using Kafka. So how to solve this problem?

Solutions are as follows:

(1) the Producer side prepares a message table and puts the two operations update DB and insert message in a DB transaction.

(2) prepare a background program to transmit the message in the message table to the message middleware continuously. Failed, keep retrying and retransmitting. Messages are allowed to repeat, but the messages are not lost and the order is not disordered.

(3) the Consumer side prepares a weight judgment table. The processed messages are recorded in the weight judgment table. Realize the idempotent of the business. But there is another atomicity problem involved here: how to guarantee the atomicity of the two operations from message consumption + insert message to judgment table?

The consumption is successful, but insert failed to judge the weight table, what should I do? This was discussed in part 1, exactly once, of Kafka's source code analysis series.

Through the above three steps, we have basically solved the atomicity of the two operations here, update db and sending network messages.

But one drawback of this solution is that it requires the design of DB message tables and a background task that constantly scans local messages. It causes the processing of messages and the coupling of business logic to increase the burden on the business side.

Scenario 2-RocketMQ transaction messages

In order to solve this problem without coupling with the business, RocketMQ put forward the concept of "transaction message".

Specifically, the sending of the message is divided into two stages: the Prepare phase and the confirmation phase.

Specifically, the above two steps are broken down into three steps:

(1) send Prepared messages

(2) update DB

(3) Confirm or cancel the Prepared message based on the success or failure of the update DB result.

Some people may ask, what if the first two steps are successful and the last one fails? Here comes the key point of RocketMQ: RocketMQ scans all Prepared messages on a regular basis (default is 1 minute) and asks the sender whether to confirm that the message is sent. Or cancel this message?

The specific code implementation is as follows:

That is, a checkListener,RocketMQ is defined to call back the Listener to implement the scheme mentioned above.

/ / as mentioned above, when RocketMQ finds the `Prepared message, it will determine the transaction TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl () according to the strategy implemented by the Listener; / / construct the producer of the transaction message TransactionMQProducer producer = new TransactionMQProducer ("groupName"); / / set the transaction decision processing class producer.setTransactionCheckListener (transactionCheckListener); / / the processing logic of the local transaction, which is equivalent to the logic of checking the Bob account and deducting money in the example TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl () Producer.start () / / Construction MSG, omitting the construction parameter Message msg = new Message (.); / / sending the message SendResult sendResult = producer.sendMessageInTransaction (msg, tranExecuter, null); producer.shutdown ()

Then execute the local transaction as follows

Public TransactionSendResult sendMessageInTransaction (.) {/ / logical code, not actual code / / 1. Send the message sendResult = this.send (msg); / / sendResult.getSendStatus () = = SEND_OK / / 2. If the message is sent successfully, the local transaction unit LocalTransactionState localTransactionState = tranExecuter.executeLocalTransactionBranch (msg, arg) associated with the message is processed; / / 3. End transaction this.endTransaction (sendResult, localTransactionState, localException);}

The listener registered on the message middleware mentioned above will call back the interface of producer after the timeout to determine the transaction execution.

Conclusion: comparing scenario 2 and solution 1, the biggest change of RocketMQ is that the business side does not do the thing of "scanning the message table", but the message middleware helps.

As for the message list, it is still not omitted. Because the message middleware asks the sender whether the execution of the thing is successful, or whether it needs a "disguised local message table" to record the execution status of the thing.

Manual intervention

Someone may want to say again, whether in scenario 1 or scenario 2, the sender successfully puts the message in the queue, but what if the consumer fails?

What if the consumption fails, try again and fail all the time? Do you want to automatically roll back the whole process?

The answer is human intervention. From the point of view of engineering practice, the cost of automatic rollback of the whole process is very huge, not only the implementation is complex, but also introduce new problems. For example, if automatic rollback fails, how to deal with it?

In response to this very low probability of case, manual processing will be more reliable and simpler than the implementation of a highly complex automatic rollback system.

The above is how to use RocketMQ transaction messages to solve distributed transactions. Have you learned any knowledge or skills? If you want to learn more skills or enrich your knowledge reserve, you are welcome to follow the industry information channel.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.