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

Introduction: Mycat has become a powerful open source distributed database middleware product. Facing the massive data transaction processing of enterprise applications, it is the best open source solution at present. However, if you want to keep the data in multiple machines consistent, a more conventional solution is to introduce a "coordinator" to schedule the execution of all nodes in a unified manner.

This paper is selected from "distributed database architecture and enterprise practice-- based on Mycat middleware".

With the concurrency, the increasing amount of data and the business has been refined so that it can no longer be divided according to the business, we have to use distributed database to improve the performance of the system. In the distributed system, each node is relatively independent physically, and the data operation on each node can meet the ACID. However, there is no way to know the execution of other node transactions among independent nodes. if you want to keep the data in multiple machines consistent, you must ensure that all data operations on all nodes are successful or not. A more conventional solution is to introduce a "coordinator" to schedule the execution of all nodes in a unified manner.

XA specification

The X/Open organization (now Open Group) defines a distributed transaction processing model. The X/Open DTP model (1994) includes four parts: application (AP), transaction manager (TM), resource manager (RM) and communication resource manager (CRM). Transaction manager (TM) is transaction middleware, resource manager (RM) is database, and communication resource manager (CRM) is message middleware. Transaction processing within a database is usually regarded as a local transaction, while the object of distributed transaction processing is a global transaction. Global transaction means that in a distributed transaction environment, multiple databases may need to work together to complete a job, which is a global transaction. Several different databases may be updated in a transaction, when a database commits its own internal operations not only for its own success, but also for other databases related to the global transaction to succeed. If any operation of any database fails, all operations done by all databases participating in this transaction must be rolled back. XA is the interface specification (interface function) between transaction middleware and database defined by X/Open DTP. Transaction middleware uses it to inform database transaction start, end, commit, rollback and so on. XA interface function is provided by database vendors. According to this idea, two-phase commit protocol and three-phase commit protocol are derived.

Two-stage submission

The so-called two phases refer to the preparation phase and the submission phase.

The preparation phase means that the transaction coordinator (transaction manager) sends a preparation message to each participant (resource manager). Each participant either directly returns a failure message (such as permission verification failure), or executes the transaction locally, writing local redo and undo logs but not committing. The preparation phase can be further divided into the following three steps.

(1) the coordinator node asks all participant nodes whether it is possible to perform a submit operation (vote) and starts waiting for a response from each participant node.

(2) the participant node performs all transaction operations until the query is initiated, and writes undo information and redo information to the log.

(3) each participant node responds to the query initiated by the coordinator node. If the participant node's transaction operation actually succeeds, it returns a "consent" message; if the participant node's transaction operation actually fails, it returns an "abort" message.

The commit phase means that if the coordinator receives a failure message from the participant or times out, a Rollback message is sent directly to each participant, otherwise a Commit message is sent, and the participant executes the commit or rollback operation according to the coordinator's instructions, releasing the lock resources used by all transactions in the process.

The shortcomings of the two-phase submission are as follows.

The main results are as follows: (1) synchronous blocking problem, in the process of execution, all participating nodes are transaction blocking type, when participants occupy common resources, other third-party nodes have to be blocked when accessing common resources.

(2) single point failure, because of the importance of the coordinator, once the coordinator fails, the participants will block all the time.

(3) the data is inconsistent. In the second stage of the two-stage submission, when a local network exception occurs after the coordinator sends the commit request to the participants or the coordinator fails in the process of sending the commit request, it will cause only some participants to receive the commit request, and this part of the participants will perform the commit operation after receiving the commit request. Other machines that do not receive commit requests are unable to perform transaction commits, so data inconsistencies occur in the whole distributed system.

Because of the defects of two-stage commit, such as synchronous blocking, single-point problem, data inconsistency, downtime and so on, the researchers made improvements on the basis of two-stage commit and proposed three-phase commit.

Three-stage submission

Three-phase commit (Three-phase commit,3PC), also known as three-phase commit protocol (Three-phase commitprotocol), is an improved version of two-phase commit (2PC). The three-phase commit divides the preparation phase of the two-phase submission into two again, so that the three-phase submission has three phases: CanCommit, PreCommit, and DoCommit.

(1) CanCommit phase: the CanCommit phase of the three-stage submission is actually very similar to the preparation phase of the two-phase submission. The coordinator sends a commit request to the participant. If the participant can submit, the participant returns the Yes response, otherwise the No response is returned.

(2) PreCommit phase: the coordinator decides whether the PreCommit operation of the transaction can be recorded according to the reaction of the participants. Depending on the response, there are two possibilities.

If the feedback the coordinator receives from all participants is an Yes response, the transaction is executed.

If any participant sends a No response to the coordinator, or if the coordinator does not receive a response from the participant after the wait timeout, the transaction is interrupted.

(3) DoCommit phase: this stage carries on the real transaction commit, and it can also be divided into two execution situations: executing commit and interrupting transaction.

The process of performing the submission is as follows.

After receiving the ACK response from the participant, the coordinator will move from the pre-submitted state to the submitted state and send the doCommit request to all participants.

After receiving the doCommit request, the transaction commit participant executes the formal transaction commit and releases all transaction resources after the transaction commit is completed.

After the transaction is committed, an ACK response is sent to the coordinator.

After receiving the ACK response from all participants, the coordinator completes the transaction. The procedure for interrupting a transaction is as follows.

The coordinator sends an abort request to all participants.

After receiving the abort request, the participant uses the undo information recorded in phase 2 to perform the rollback operation of the transaction, and releases all transaction resources after the rollback is completed.

After the participant completes the transaction rollback, an ACK message is sent to the coordinator.

After the coordinator receives the ACK message from the participant, the transaction is interrupted.

Implementation of distributed transaction in Mycat

Mycat has fully supported the XA distributed strong transaction type since version 1.6, so let's first use a simple example to understand the use of XA in Mycat.

The user application side (AP) is used as follows:

(1) set autocommit=0

Transactions need to be set up in the application layer that cannot be committed automatically

(2) set xa=on

Set XA to on in SQL

(3) execute SQL

Insert into travelrecord (id,name) values (1 recorder N'), (60000 dh), (321 recorder D'), (13400000 dh), (59m e')

(4) commit or rollback

Commit the transaction (commit successfully or rollback exception).

The complete flow chart is shown in the figure.

The implementation flow of the internal implementation side of Mycat is as follows:

(1) set autocommit=0

Set autocommit in MysqlConnection to false

(2) set xa=on

Open the XA transaction manager in Mycat, generate the XID with MycatServer.getInstance (). GenXATXID (), mark the beginning of the XA transaction with the XA START XID command, continue to assemble the SQL business (Mycat will fragment the above insert data to different nodes), assemble the XA END XID,XA PREPARE XID and finally commit the 1pc and log it to the tm.log. If there is an exception in the 1pc phase, roll back the transaction XA ROLLBACK xid directly.

(3) all 2pc commit (XA COMMIT) is performed in multi-node MySQL. After the commit is successful, the transaction ends; if there is an exception, the transaction is resubmitted or rolled back.

The exception handling flow for XA distributed transactions in Mycat is as follows:

(1) one-stage commit exception: if 1pc commits any mysql node cannot commit or an exception, the transactions of all nodes are rolled back, and an exception is thrown to the application side transaction rollback.

(2) Mycat Crash Recovery

After the Mycat crashes, restart and recover according to the tm.log transaction log. After mycat is started, the transaction log is executed to find the XA transactions that have been prepared in each node, and then commit or rollback is performed.

1. Related class description

Send set xa = on through the user's application side; SQL enables the function of the XA transaction manager within Mycat, and the transaction manager will manage the MySQL database in XA mode. The implementation code of the transaction management function is as follows:

MySQLConnection: database connection.

NonBlockingSession: the user connects to the Session.

MultiNodeCoordinator: coordinator.

CommitNodeHandler: multipart submission processing.

RollbackNodeHandler: multipart rollback processing.

two。 Code parsing

The source code for starting a XA transaction is as follows:

Public class MySQLConnection extends BackendAIOConnection {/ / set open transaction private void getAutocommitCommand (StringBuilder sb, boolean autoCommit) {if (autoCommit) {sb.append ("SET autocommit=1;");} else {sb.append ("SET autocommit=0;") }} public void execute (RouteResultsetNode rrn, ServerConnection sc,boolean autocommit) throws UnsupportedEncodingException {if (! modifiedSQLExecuted & & rrn.isModifySQL ()) {modifiedSQLExecuted = true;} / / get the current transaction ID String xaTXID = sc.getSession2 (). GetXaTXID (); synAndDoExecute (xaTXID, rrn, sc.getCharsetIndex (), sc.getTxIsolation (), autocommit);}. / / the code here is omitted. Readers are advised to refer to the MySQLConnection.java source code of the MyCAT-Server project in GitHub warehouse.

After the user's application side settings are manually submitted, Mycat will be added to the current connection

SET autocommit=0

Add this statement to the StringBuffer and wait for it to be submitted to the database.

The source code for a user to connect to Session is as follows:

Public class NonBlockingSession implements Session {. / / the code here is omitted. Readers are advised to refer to the NonBlockingSession.java source code of the MyCAT-Server project in GitHub warehouse} SET XA = ON; sentence analysis

The user's application side sends the statement to Mycat, which is parsed by the SQL statement parser and handed over to SetHandle to process c.getSession2 (). SetXATXEnabled (true)

Call the setXATXEnable d method in NonBlockSession to set the XA switch to start, and generate XID, as follows:

Public void setXATXEnabled (boolean xaTXEnabled) {LOGGER.info ("XA Transaction enabled, con" + this.getSource ()); if (xaTXEnabled & & this.xaTXID = = null) {xaTXID = genXATXID ();}}

In addition, the NonBlockSession receives the commit from the user's application side and calls the commit method to handle the transaction commit logic.

In the commit () method, we will first check the number of nodes. One node and multiple nodes are divided into different processes. Here we only talk about the processing method checkDistriTransaxAndExecute () of multiple nodes.

This method commits transactions from multiple nodes.

The source code of the coordinator is as follows:

Public class MultiNodeCoordinator implements ResponseHandler {. / / the code here is omitted. Readers are advised to refer to the MultiNodeCoordinator.java source code of the GitHub warehouse MyCAT-Server project.

In the checkDistriTransaxAndExecute () method of NonBlockSession, the NonBlockSession session class calls the MultiNodeCoordinator class that specializes in multi-node cooperation for specific processing. In the MultiNodeCoordinator class, the executeBatchNodeCmd method adds the processing of XA 1PC submission. The code snippet is as follows:

For (RouteResultsetNode rrn: session.getTargetKeys ()) {. If (mysqlCon.getXaStatus () = = TxState.TX_STARTED_STATE) {/ / recovery Log participantLogEntry [started] = new ParticipantLogEntry (xaTxId,conn.getHost (), 0cmds conn.getSchema (), ((MySQLConnection) conn) .getXaStatus ()); String [] cmds = new String [] {"XA END" + xaTxId, "XA PREPARE" + xaTxId} If (LOGGER.isDebugEnabled ()) {LOGGER.debug ("Start execute the batch cmd:" + cmds [0] + ";" + cmds [1] + "," + "current connection:" + conn.getHost () + ":" + conn.getPort ());} mysqlCon.execBatchCmd (cmds);}. }

In the okResponse method of the MultiNodeCoordinator class, the transaction commit of the 2pc is performed

MySQLConnection mysqlCon = (MySQLConnection) conn;switch (mysqlCon.getXaStatus ()) {case TxState.TX_STARTED_STATE: if (mysqlCon.batchCmdFinished ()) {String xaTxId = session.getXaTXID (); String cmd = "XA COMMIT" + xaTxId; if (LOGGER.isDebugEnabled ()) {LOGGER.debug ("Start execute the cmd:" + cmd+ ", current host:" + mysqlCon.getHost () + ":" + mysqlCon.getPort ()) } / / recovery log CoordinatorLogEntry coordinatorLogEntry = inMemoryRepository.get (xaTxId); for (int iTuno; I

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.