Six schemes to ensure data consistency of distributed system

Editor's Note: This article is organized by a group discussion on "High Availability Architecture Back Garden".

Where there are people, there are rivers and lakes.

Where there are rivers and lakes, there are disputes.

In e-commerce and other businesses, the system is generally composed of multiple independent services. How to solve the consistency of data in distributed call?

The specific business scenarios are as follows. For example, a business operation calls services A, B and C at the same time, which need to be satisfied or successful at the same time; Or failures A, B and C may be remote services developed by different departments and deployed on different servers.

In the distributed system, if we don't want to sacrifice consistency, CAP theory tells us that we can only give up availability, which is obviously unacceptable. In order to facilitate the discussion, the basic theory of data consistency is briefly introduced at first.

Strong consistency

Weak consistency

Final consistency

In engineering practice, in order to ensure the availability of the system, most Internet systems convert the requirement of strong consistency into the requirement of final consistency, and ensure the final consistency of data by ensuring the idempotency of the system. However, in e-commerce and other scenarios, the solution of data consistency is different from the common Internet system (such as MySQL master-slave synchronization), and the group discussion is divided into the following six solutions.

The business integration scheme mainly adopts the method of interface integration to local execution. For example, in a problem scenario, services A, B and C can be integrated into a service D for business, and this service D can be converted into a local transaction. For example, service D includes local service and service E, which is the integration of local services A ~ C. ..

Advantages: Distributed transactions are solved (circumvented).

Disadvantages: Obviously, the services originally planned to be split are coupled together, and the business responsibilities are not clear, which is not conducive to maintenance.

Because this method has obvious shortcomings, it is usually not recommended.

The core of this scheme is to execute tasks that need distributed processing asynchronously through message logs. Message log can be stored in local text, database or message queue, and then retry can be initiated automatically or manually through business rules. Manual retry is more used in payment scenarios, and problems afterwards are handled through the reconciliation system.

The core of message logging scheme is to ensure the idempotency of service interface.

Considering the network communication failure, packet loss and other reasons, if the interface can not guarantee idempotency, it is difficult to guarantee the uniqueness of data.

Yi Bei's main points are as follows.

Alkali: a substitute for acid

This project is an article published by Yi Bei architect Dan pritchett to ACM in 2008. This is a classic article explaining the basic principle or ultimate consistency. This paper discusses the basic differences between basic principles and acidic principles in ensuring data consistency.

If ACID provides a consistent choice for partitioned databases, how can availability be achieved? The answer is

Basic (basically available, soft state, finally consistent)

Basic availability is achieved by supporting local failures rather than global system failures. Here is a simple example: if users are divided into five database servers, the basic design encourages similar processing, and the failure of a user database only affects 20% users of this particular host. There is no magic involved here, but it does bring higher usability of the perception system.

This article describes one of the most common scenarios. If a transaction is generated, you need to add records to the transaction table and modify the amount in the user table. These two tables belong to different remote services, so it involves the problem of distributed transaction consistency.

This paper presents a classic solution, which puts the main modification operation and the message of updating the user table in a local transaction. At the same time, in order to avoid the problems caused by repeated consumption of user table messages and realize the idempotency of multiple retries, the update record table updates_applied is added to record the processed messages.

The pseudo code of the system is as follows

(Click to zoom the picture in full screen)

Based on the above method, in the first stage, the transaction table and message queue are added through the transaction guarantee of the local database.

In the second stage, the message queue is read out independently (but not deleted), and whether the related records are executed is detected by judging the update record table updates_applied. The unexecuted record will modify the user table, then add an operation record in updates_applied, and delete the queue after the transaction is successfully executed.

Through the above methods, the final consistency of the distributed system is realized. For more information about the Yi Bei project, please refer to the link at the end of the article.

With the continuous expansion of business scale, e-commerce websites generally have to face the road of splitting. Is to split the original single application into several subsystems with different responsibilities. For example, the functions that used to face users, customers and operations may be put in one system, but now they are divided into several subsystems, such as order center, agent management, operation system, quotation center and inventory management.

What is the first thing to face in division?

At the beginning, all the functions of a single application were together and stored together. For example, to cancel the order, directly update the status of the order table, and then update the inventory table. Because it is a single application and stored together, all this can be done in one transaction, and the relational database can ensure consistency.

But after the split, it's different. Different subsystems have their own storage. For example, the order center only manages its own order library, and inventory management also has its own library. Then when the operating system cancels the order, it calls the service of the order center and inventory management through the interface call, instead of directly operating the library. This involves a "distributed transaction" problem.

There are two solutions to distributed transactions.

1. Use asynchronous messages first.

As mentioned above, when using asynchronous message consumers, idempotency needs to be realized.

There are two ways to idempotent. One way is that business logic guarantees idempotency. For example, after receiving the message of successful payment, the order status will change to payment completed. If the current status is payment completed, receiving another message of successful payment means that the message is repeated and will be directly treated as a message.

In other words, if business logic cannot guarantee idempotency, a deduplication table or similar implementation should be added. For the producer, the message base is placed on the same instance of the business database, and messages and business operations are sent in the same local transaction. When sending a message, the message is not sent immediately, but a message record is inserted in the message base, and then the message is sent asynchronously when the transaction is committed. If the message is sent successfully, the message in the message library will be deleted. If the message queuing service is abnormal or there is a network problem and the message is not sent successfully, then the message will stay here, and another service will constantly scan out these messages and resend them.

2. Some services are not suitable for asynchronous messaging, and all participants in the transaction need to get the results synchronously. In fact, the implementation of this situation is similar to the above, and there is a transaction record library on the same instance of each participant's local business library.

For example, a calls b and c synchronously. A updates the local transaction status when the local transaction is successful, and B and C are the same. If A fails to call B once, it may be that B really failed, or it may be that the call timed out, and B actually succeeded. Then a central service compares the transaction records of the three parties and makes a final decision. Suppose that the transaction records of the three parties now are A success, B failure and C success. Then there are two ways to finally decide, according to the specific scenario:

Give a special explanation to Scenario B: For example, B is an inventory deduction service, which failed for some reason when it was first called, but the inventory has become 0 when it is retried, so it is impossible to retry successfully. At this point, A and C must be rolled back.

Then some people may think that putting a message library or transaction record library in the same instance of the business library will infringe on the business, and the business should care about this library. Is the design reasonable?

In fact, we can rely on operation and maintenance to simplify the intrusion of development. Our method is to let DBA initialize this library in advance on all MySQL instances of the company, and operate this library transparently in the background through the framework layer (message client or transaction RPC framework). Business developers only need to care about their own business logic and do not need to directly access this library.

To sum up, in fact, the basic principles of the two methods are similar, both of which are to transform distributed transactions into multiple local transactions, and then rely on retry to achieve the final consistency.

General process of transaction creation

We abstract the transaction creation process into a series of extensible function points, and each function point can have multiple implementations (there is a combination/mutual exclusion relationship between the specific implementations). String all the function points together according to a certain process, and the process of transaction creation is completed.

Problems faced

The implementation of each function point may depend on external services. So how to ensure the data consistency between services? For example, the call to lock the coupon service timed out, and it is uncertain whether the coupon was locked successfully. What should I do? For another example, the ticket was locked successfully, but the deduction of inventory failed. What should I do?

Scheme selection

Excessive service dependence will lead to the problems of increased management complexity and increased stability risk. Imagine, if we rely on 10 services, and all 9 services succeed and the last one fails, will the first 9 services be rolled back? This cost is still very high.

Therefore, on the premise of splitting the large process into several small local transactions, we choose the scheme of sending message notification and executing related transactions asynchronously after the local transactions are successfully executed, in view of the non-real-time and non-strong consistency of related business writing.

Message notification often cannot guarantee 100% success; And it is still unknown whether the service of the receiver can be successfully executed after the message is notified. The former problem can be solved by retrying; The latter can be guaranteed by transaction messages.

So at present, only business scenarios need to be synchronized in real time, which has strong consistency requirements. In the process of transaction creation, two typical scenarios are securities locking and inventory deduction.

To ensure data consistency among multiple systems, at first glance, a distributed transaction framework needs to be introduced to solve it. However, the introduction of a very heavy distributed transaction framework similar to two-phase commit will bring about a sharp increase in complexity; In the field of e-commerce, absolute strong consistency is too idealistic, so we can choose quasi-real-time final consistency.

In the process of transaction creation, we first create an invisible order, and then send a scrap message to MQ for the abnormal call (failure or timeout) when the securities are locked and the inventory is deducted synchronously. If the message fails to be sent, time stepping asynchronous retry will be performed locally; After receiving the message, the coupon system and the inventory system will judge whether it is necessary to do business rollback, so as to ensure the final consistency of multiple local transactions in quasi-real time.

There is also a commonly used xts scheme of Alipay, which is improved by Alipay on the basis of 2PC. The main idea is as follows. Most of the information is quoted from official website.

Brief introduction of distributed transaction service

Distributed Transaction Service (DTS) is a distributed transaction framework, which is used to ensure the final consistency of transactions in a large-scale distributed environment. DTS is divided into xts client and xts server in architecture. The former is a JAR package embedded in the client application, which is mainly responsible for writing and processing transaction data. The latter is an independent system, mainly responsible for the recovery of abnormal transactions.

Core characteristics

The transaction model of traditional relational database must abide by the ACID principle. In the single database mode, the ACID model can effectively ensure the integrity of data, but in a large-scale distributed environment, a business often spans multiple databases. How to ensure data consistency among these databases requires other effective strategies. In JavaEE specification, 2PC (2 Phase Commit) is used to deal with the transaction problem in cross-DB environment, but 2PC is an anti-scalable mode, that is, during the transaction processing, participants need to hold resources until the end of the whole distributed transaction. In this way, when the business scale reaches the level of10 million, the limitations of 2PC will become more and more obvious, and the system scalability will become very poor. Based on this, we adopt the idea of BASE to realize a set of distributed transaction scheme similar to 2PC, that is, DTS. DTS fully ensures the high availability and reliability in the distributed environment, while taking into account the requirements of data consistency. Its biggest feature is to ensure the final consistency of data.

Simply put, the DTS framework has the following characteristics:

The following is a flow chart of the distributed transaction framework.

realize

Compared with 2PC protocol

1. E-commerce business

The payment department of the company provides payment services for the business department by accessing other third-party payment systems. Payment service is an RPC service based on Dubbo.

For the business department, it is necessary to call the order payment of the e-commerce department.

From the business rules, it is necessary to ensure the real-time and consistency of business data at the same time, that is, the success of payment must be rewarded.

The way we adopt is to call synchronously, and deal with local transactions first. Considering that the integral business is relatively simple and the business impact is lower than payment, the integral platform provides an interface for increase and decrease.

The specific process is to call the integration platform to increase user points, and then call the payment platform for payment processing. If the processing fails, the catch method calls the cancellation method of the integration platform to cancel the integral order processed this time.

(Click on the picture to enlarge it in full screen)

2. User information change

Distributed services need more supporting systems, especially our final consistency scheme based on messages and logs, which needs to consider the backlog, consumption, monitoring and alarm of messages.

In partitioned databases, sacrificing some consistency for availability can greatly improve scalability.

English version: http://queue.acm.org/detail.cfm? id = 1394 128

Chinese version: http://article.yeeyan.org/view/167444/125572.

I am grateful to Li Yufu, Yu, and Qigong of Mushroom Street for their suggestions, and many other team members for their contributions to the content of this article.

This article was edited by Li Yufu and Tim Yang. Please indicate from @ High Availability Architecture.