乐观事务和悲观事务

简单的讲，乐观事务模型就是直接提交，遇到冲突就回滚，悲观事务模型就是在真正提交事务前，先尝试对需要修改的资源上锁，只有在确保事务一定能够执行成功后，才开始提交。

对于乐观事务模型来说，比较适合冲突率不高的场景，因为直接提交大概率会成功，冲突是小概率事件，但是一旦遇到事务冲突，回滚的代价会比较大。

悲观事务的好处是对于冲突率高的场景，提前上锁的代价小于事后回滚的代价，而且还能以比较低的代价解决多个并发事务互相冲突导致谁也成功不了的场景。不过悲观事务在冲突率不高的场景并没有乐观事务处理高效。

从应用端实现的复杂度而言，悲观事务更直观，更容易实现。而乐观事务需要复杂的应用端重试机制来保证。

下面用 bookshop 数据库中的表实现一个购书的例子来演示乐观事务和悲观事务的区别以及优缺点。购书流程主要包括：

  1. 更新库存数量
  2. 创建订单
  3. 付款

这三个操作需要保证全部成功或者全部失败，并且在并发情况下要保证不超卖。

悲观事务

下面代码以悲观事务的方式，用两个线程模拟了两个用户并发买同一本书的过程，书店剩余 10 本，Bob 购买了 6 本，Alice 购买了 4 本。两个人几乎同一时间完成订单，最终，这本书的剩余库存为零。

  • Java
  • Golang
  • Python

当使用多个线程模拟多用户同时插入的情况时，需要使用一个线程安全的连接对象，这里使用 Java 当前较流行的连接池 HikariCP。

Golang 的 sql.DB 是并发安全的，无需引入外部包。

封装一个用于适配 TiDB 事务的工具包 util，编写以下代码备用：

  1. package util
  3. import (
  4.     "context"
  5.     "database/sql"
  6. )
  8. type TiDBSqlTx struct {
  9.     *sql.Tx
  10.     conn        *sql.Conn
  11.     pessimistic bool
  12. }
  14. func TiDBSqlBegin(db *sql.DB, pessimistic bool) (*TiDBSqlTx, error) {
  15.     ctx := context.Background()
  16.     conn, err := db.Conn(ctx)
  17.     if err != nil {
  18.         return nil, err
  19.     }
  20.     if pessimistic {
  21.         _, err = conn.ExecContext(ctx, "set @@tidb_txn_mode=?", "pessimistic")
  22.     } else {
  23.         _, err = conn.ExecContext(ctx, "set @@tidb_txn_mode=?", "optimistic")
  24.     }
  25.     if err != nil {
  26.         return nil, err
  27.     }
  28.     tx, err := conn.BeginTx(ctx, nil)
  29.     if err != nil {
  30.         return nil, err
  31.     }
  32.     return &TiDBSqlTx{
  33.         conn:        conn,
  34.         Tx:          tx,
  35.         pessimistic: pessimistic,
  36.     }, nil
  37. }
  39. func (tx *TiDBSqlTx) Commit() error {
  40.     defer tx.conn.Close()
  41.     return tx.Tx.Commit()
  42. }
  44. func (tx *TiDBSqlTx) Rollback() error {
  45.     defer tx.conn.Close()
  46.     return tx.Tx.Rollback()
  47. }

使用 Python 的 mysqlclient Driver 开启多个连接对象进行交互，线程之间不共享连接，以保证其线程安全。

1. 编写悲观事务示例

  • Java
  • Golang
  • Python

配置文件

在 Java 中，如果你使用 Maven 作为包管理，在 pom.xml 中的 <dependencies> 节点中，加入以下依赖来引入 HikariCP，同时设定打包目标，及 JAR 包启动的主类，完整的 pom.xml 如下所示:

  1. <?xml version="1.0" encoding="UTF-8"?>
  3. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.   <modelVersion>4.0.0</modelVersion>
  7.   <groupId>com.pingcap</groupId>
  8.   <artifactId>plain-java-txn</artifactId>
  9.   <version>0.0.1</version>
  11.   <name>plain-java-jdbc</name>
  13.   <properties>
  14.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  15.     <maven.compiler.source>17</maven.compiler.source>
  16.     <maven.compiler.target>17</maven.compiler.target>
  17.   </properties>
  19.   <dependencies>
  20.     <dependency>
  21.       <groupId>junit</groupId>
  22.       <artifactId>junit</artifactId>
  23.       <version>4.13.2</version>
  24.       <scope>test</scope>
  25.     </dependency>
  27.     <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
  28.     <dependency>
  29.       <groupId>mysql</groupId>
  30.       <artifactId>mysql-connector-java</artifactId>
  31.       <version>8.0.28</version>
  32.     </dependency>
  34.     <dependency>
  35.       <groupId>com.zaxxer</groupId>
  36.       <artifactId>HikariCP</artifactId>
  37.       <version>5.0.1</version>
  38.     </dependency>
  39.   </dependencies>
  41.   <build>
  42.     <plugins>
  43.       <plugin>
  44.         <groupId>org.apache.maven.plugins</groupId>
  45.         <artifactId>maven-assembly-plugin</artifactId>
  46.         <version>3.3.0</version>
  47.         <configuration>
  48.           <descriptorRefs>
  49.             <descriptorRef>jar-with-dependencies</descriptorRef>
  50.           </descriptorRefs>
  51.           <archive>
  52.             <manifest>
  53.               <mainClass>com.pingcap.txn.TxnExample</mainClass>
  54.             </manifest>
  55.           </archive>
  57.         </configuration>
  58.         <executions>
  59.           <execution>
  60.             <id>make-assembly</id>
  61.             <phase>package</phase>
  62.             <goals>
  63.               <goal>single</goal>
  64.             </goals>
  65.           </execution>
  66.         </executions>
  67.       </plugin>
  68.     </plugins>
  69.   </build>
  71. </project>

代码

随后编写代码：

  1. package com.pingcap.txn;
  3. import com.zaxxer.hikari.HikariDataSource;
  5. import java.math.BigDecimal;
  6. import java.sql.*;
  7. import java.util.Arrays;
  8. import java.util.concurrent.*;
  10. public class TxnExample {
  11.     public static void main(String[] args) throws SQLException, InterruptedException {
  12.         System.out.println(Arrays.toString(args));
  13.         int aliceQuantity = 0;
  14.         int bobQuantity = 0;
  16.         for (String arg: args) {
  17.             if (arg.startsWith("ALICE_NUM")) {
  18.                 aliceQuantity = Integer.parseInt(arg.replace("ALICE_NUM=", ""));
  19.             }
  21.             if (arg.startsWith("BOB_NUM")) {
  22.                 bobQuantity = Integer.parseInt(arg.replace("BOB_NUM=", ""));
  23.             }
  24.         }
  26.         HikariDataSource ds = new HikariDataSource();
  27.         ds.setJdbcUrl("jdbc:mysql://localhost:4000/bookshop?useServerPrepStmts=true&cachePrepStmts=true");
  28.         ds.setUsername("root");
  29.         ds.setPassword("");
  31.         // prepare data
  32.         Connection connection = ds.getConnection();
  33.         createBook(connection, 1L, "Designing Data-Intensive Application", "Science & Technology",
  34.                 Timestamp.valueOf("2018-09-01 00:00:00"), new BigDecimal(100), 10);
  35.         createUser(connection, 1L, "Bob", new BigDecimal(10000));
  36.         createUser(connection, 2L, "Alice", new BigDecimal(10000));
  38.         CountDownLatch countDownLatch = new CountDownLatch(2);
  39.         ExecutorService threadPool = Executors.newFixedThreadPool(2);
  41.         final int finalBobQuantity = bobQuantity;
  42.         threadPool.execute(() -> {
  43.             buy(ds, 1, 1000L, 1L, 1L, finalBobQuantity);
  44.             countDownLatch.countDown();
  45.         });
  46.         final int finalAliceQuantity = aliceQuantity;
  47.         threadPool.execute(() -> {
  48.             buy(ds, 2, 1001L, 1L, 2L, finalAliceQuantity);
  49.             countDownLatch.countDown();
  50.         });
  52.         countDownLatch.await(5, TimeUnit.SECONDS);
  53.     }
  55.     public static void createUser(Connection connection, Long id, String nickname, BigDecimal balance) throws SQLException  {
  56.         PreparedStatement insert = connection.prepareStatement(
  57.                 "INSERT INTO `users` (`id`, `nickname`, `balance`) VALUES (?, ?, ?)");
  58.         insert.setLong(1, id);
  59.         insert.setString(2, nickname);
  60.         insert.setBigDecimal(3, balance);
  61.         insert.executeUpdate();
  62.     }
  64.     public static void createBook(Connection connection, Long id, String title, String type, Timestamp publishedAt, BigDecimal price, Integer stock) throws SQLException {
  65.         PreparedStatement insert = connection.prepareStatement(
  66.                 "INSERT INTO `books` (`id`, `title`, `type`, `published_at`, `price`, `stock`) values (?, ?, ?, ?, ?, ?)");
  67.         insert.setLong(1, id);
  68.         insert.setString(2, title);
  69.         insert.setString(3, type);
  70.         insert.setTimestamp(4, publishedAt);
  71.         insert.setBigDecimal(5, price);
  72.         insert.setInt(6, stock);
  74.         insert.executeUpdate();
  75.     }
  77.     public static void buy (HikariDataSource ds, Integer threadID,
  78.                             Long orderID, Long bookID, Long userID, Integer quantity) {
  79.         String txnComment = "/* txn " + threadID + " */ ";
  81.         try (Connection connection = ds.getConnection()) {
  82.             try {
  83.                 connection.setAutoCommit(false);
  84.                 connection.createStatement().executeUpdate(txnComment + "begin pessimistic");
  86.                 // waiting for other thread ran the 'begin pessimistic' statement
  87.                 TimeUnit.SECONDS.sleep(1);
  89.                 BigDecimal price = null;
  91.                 // read price of book
  92.                 PreparedStatement selectBook = connection.prepareStatement(txnComment + "select price from books where id = ? for update");
  93.                 selectBook.setLong(1, bookID);
  94.                 ResultSet res = selectBook.executeQuery();
  95.                 if (!res.next()) {
  96.                     throw new RuntimeException("book not exist");
  97.                 } else {
  98.                     price = res.getBigDecimal("price");
  99.                 }
  101.                 // update book
  102.                 String updateBookSQL = "update `books` set stock = stock - ? where id = ? and stock - ? >= 0";
  103.                 PreparedStatement updateBook = connection.prepareStatement(txnComment + updateBookSQL);
  104.                 updateBook.setInt(1, quantity);
  105.                 updateBook.setLong(2, bookID);
  106.                 updateBook.setInt(3, quantity);
  107.                 int affectedRows = updateBook.executeUpdate();
  109.                 if (affectedRows == 0) {
  110.                     // stock not enough, rollback
  111.                     connection.createStatement().executeUpdate(txnComment + "rollback");
  112.                     return;
  113.                 }
  115.                 // insert order
  116.                 String insertOrderSQL = "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (?, ?, ?, ?)";
  117.                 PreparedStatement insertOrder = connection.prepareStatement(txnComment + insertOrderSQL);
  118.                 insertOrder.setLong(1, orderID);
  119.                 insertOrder.setLong(2, bookID);
  120.                 insertOrder.setLong(3, userID);
  121.                 insertOrder.setInt(4, quantity);
  122.                 insertOrder.executeUpdate();
  124.                 // update user
  125.                 String updateUserSQL = "update `users` set `balance` = `balance` - ? where id = ?";
  126.                 PreparedStatement updateUser = connection.prepareStatement(txnComment + updateUserSQL);
  127.                 updateUser.setBigDecimal(1, price.multiply(new BigDecimal(quantity)));
  128.                 updateUser.setLong(2, userID);
  129.                 updateUser.executeUpdate();
  131.                 connection.createStatement().executeUpdate(txnComment + "commit");
  132.             } catch (Exception e) {
  133.                 connection.createStatement().executeUpdate(txnComment + "rollback");
  134.                 e.printStackTrace();
  135.             }
  136.         } catch (SQLException e) {
  137.             e.printStackTrace();
  138.         }
  139.     }
  140. }

首先编写一个封装了所需的数据库操作的 helper.go 文件：

  1. package main
  3. import (
  4.     "context"
  5.     "database/sql"
  6.     "fmt"
  7.     "time"
  9.     "github.com/go-sql-driver/mysql"
  10.     "github.com/pingcap-inc/tidb-example-golang/util"
  11.     "github.com/shopspring/decimal"
  12. )
  14. type TxnFunc func(txn *util.TiDBSqlTx) error
  16. const (
  17.     ErrWriteConflict      = 9007 // Transactions in TiKV encounter write conflicts.
  18.     ErrInfoSchemaChanged  = 8028 // table schema changes
  19.     ErrForUpdateCantRetry = 8002 // "SELECT FOR UPDATE" commit conflict
  20.     ErrTxnRetryable       = 8022 // The transaction commit fails and has been rolled back
  21. )
  23. const retryTimes = 5
  25. var retryErrorCodeSet = map[uint16]interface{}{
  26.     ErrWriteConflict:      nil,
  27.     ErrInfoSchemaChanged:  nil,
  28.     ErrForUpdateCantRetry: nil,
  29.     ErrTxnRetryable:       nil,
  30. }
  32. func runTxn(db *sql.DB, optimistic bool, optimisticRetryTimes int, txnFunc TxnFunc) {
  33.     txn, err := util.TiDBSqlBegin(db, !optimistic)
  34.     if err != nil {
  35.         panic(err)
  36.     }
  38.     err = txnFunc(txn)
  39.     if err != nil {
  40.         txn.Rollback()
  41.         if mysqlErr, ok := err.(*mysql.MySQLError); ok && optimistic && optimisticRetryTimes != 0 {
  42.             if _, retryableError := retryErrorCodeSet[mysqlErr.Number]; retryableError {
  43.                 fmt.Printf("[runTxn] got a retryable error, rest time: %d\n", optimisticRetryTimes-1)
  44.                 runTxn(db, optimistic, optimisticRetryTimes-1, txnFunc)
  45.                 return
  46.             }
  47.         }
  49.         fmt.Printf("[runTxn] got an error, rollback: %+v\n", err)
  50.     } else {
  51.         err = txn.Commit()
  52.         if mysqlErr, ok := err.(*mysql.MySQLError); ok && optimistic && optimisticRetryTimes != 0 {
  53.             if _, retryableError := retryErrorCodeSet[mysqlErr.Number]; retryableError {
  54.                 fmt.Printf("[runTxn] got a retryable error, rest time: %d\n", optimisticRetryTimes-1)
  55.                 runTxn(db, optimistic, optimisticRetryTimes-1, txnFunc)
  56.                 return
  57.             }
  58.         }
  60.         if err == nil {
  61.             fmt.Println("[runTxn] commit success")
  62.         }
  63.     }
  64. }
  66. func prepareData(db *sql.DB, optimistic bool) {
  67.     runTxn(db, optimistic, retryTimes, func(txn *util.TiDBSqlTx) error {
  68.         publishedAt, err := time.Parse("2006-01-02 15:04:05", "2018-09-01 00:00:00")
  69.         if err != nil {
  70.             return err
  71.         }
  73.         if err = createBook(txn, 1, "Designing Data-Intensive Application",
  74.             "Science & Technology", publishedAt, decimal.NewFromInt(100), 10); err != nil {
  75.             return err
  76.         }
  78.         if err = createUser(txn, 1, "Bob", decimal.NewFromInt(10000)); err != nil {
  79.             return err
  80.         }
  82.         if err = createUser(txn, 2, "Alice", decimal.NewFromInt(10000)); err != nil {
  83.             return err
  84.         }
  86.         return nil
  87.     })
  88. }
  90. func buyPessimistic(db *sql.DB, goroutineID, orderID, bookID, userID, amount int) {
  91.     txnComment := fmt.Sprintf("/* txn %d */ ", goroutineID)
  92.     if goroutineID != 1 {
  93.         txnComment = "\t" + txnComment
  94.     }
  96.     fmt.Printf("\nuser %d try to buy %d books(id: %d)\n", userID, amount, bookID)
  98.     runTxn(db, false, retryTimes, func(txn *util.TiDBSqlTx) error {
  99.         time.Sleep(time.Second)
  101.         // read the price of book
  102.         selectBookForUpdate := "select `price` from books where id = ? for update"
  103.         bookRows, err := txn.Query(selectBookForUpdate, bookID)
  104.         if err != nil {
  105.             return err
  106.         }
  107.         fmt.Println(txnComment + selectBookForUpdate + " successful")
  108.         defer bookRows.Close()
  110.         price := decimal.NewFromInt(0)
  111.         if bookRows.Next() {
  112.             err = bookRows.Scan(&price)
  113.             if err != nil {
  114.                 return err
  115.             }
  116.         } else {
  117.             return fmt.Errorf("book ID not exist")
  118.         }
  119.         bookRows.Close()
  121.         // update book
  122.         updateStock := "update `books` set stock = stock - ? where id = ? and stock - ? >= 0"
  123.         result, err := txn.Exec(updateStock, amount, bookID, amount)
  124.         if err != nil {
  125.             return err
  126.         }
  127.         fmt.Println(txnComment + updateStock + " successful")
  129.         affected, err := result.RowsAffected()
  130.         if err != nil {
  131.             return err
  132.         }
  134.         if affected == 0 {
  135.             return fmt.Errorf("stock not enough, rollback")
  136.         }
  138.         // insert order
  139.         insertOrder := "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (?, ?, ?, ?)"
  140.         if _, err := txn.Exec(insertOrder,
  141.             orderID, bookID, userID, amount); err != nil {
  142.             return err
  143.         }
  144.         fmt.Println(txnComment + insertOrder + " successful")
  146.         // update user
  147.         updateUser := "update `users` set `balance` = `balance` - ? where id = ?"
  148.         if _, err := txn.Exec(updateUser,
  149.             price.Mul(decimal.NewFromInt(int64(amount))), userID); err != nil {
  150.             return err
  151.         }
  152.         fmt.Println(txnComment + updateUser + " successful")
  154.         return nil
  155.     })
  156. }
  158. func buyOptimistic(db *sql.DB, goroutineID, orderID, bookID, userID, amount int) {
  159.     txnComment := fmt.Sprintf("/* txn %d */ ", goroutineID)
  160.     if goroutineID != 1 {
  161.         txnComment = "\t" + txnComment
  162.     }
  164.     fmt.Printf("\nuser %d try to buy %d books(id: %d)\n", userID, amount, bookID)
  166.     runTxn(db, true, retryTimes, func(txn *util.TiDBSqlTx) error {
  167.         time.Sleep(time.Second)
  169.         // read the price and stock of book
  170.         selectBookForUpdate := "select `price`, `stock` from books where id = ? for update"
  171.         bookRows, err := txn.Query(selectBookForUpdate, bookID)
  172.         if err != nil {
  173.             return err
  174.         }
  175.         fmt.Println(txnComment + selectBookForUpdate + " successful")
  176.         defer bookRows.Close()
  178.         price, stock := decimal.NewFromInt(0), 0
  179.         if bookRows.Next() {
  180.             err = bookRows.Scan(&price, &stock)
  181.             if err != nil {
  182.                 return err
  183.             }
  184.         } else {
  185.             return fmt.Errorf("book ID not exist")
  186.         }
  187.         bookRows.Close()
  189.         if stock < amount {
  190.             return fmt.Errorf("book not enough")
  191.         }
  193.         // update book
  194.         updateStock := "update `books` set stock = stock - ? where id = ? and stock - ? >= 0"
  195.         result, err := txn.Exec(updateStock, amount, bookID, amount)
  196.         if err != nil {
  197.             return err
  198.         }
  199.         fmt.Println(txnComment + updateStock + " successful")
  201.         affected, err := result.RowsAffected()
  202.         if err != nil {
  203.             return err
  204.         }
  206.         if affected == 0 {
  207.             return fmt.Errorf("stock not enough, rollback")
  208.         }
  210.         // insert order
  211.         insertOrder := "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (?, ?, ?, ?)"
  212.         if _, err := txn.Exec(insertOrder,
  213.             orderID, bookID, userID, amount); err != nil {
  214.             return err
  215.         }
  216.         fmt.Println(txnComment + insertOrder + " successful")
  218.         // update user
  219.         updateUser := "update `users` set `balance` = `balance` - ? where id = ?"
  220.         if _, err := txn.Exec(updateUser,
  221.             price.Mul(decimal.NewFromInt(int64(amount))), userID); err != nil {
  222.             return err
  223.         }
  224.         fmt.Println(txnComment + updateUser + " successful")
  226.         return nil
  227.     })
  228. }
  230. func createBook(txn *util.TiDBSqlTx, id int, title, bookType string,
  231.     publishedAt time.Time, price decimal.Decimal, stock int) error {
  232.     _, err := txn.ExecContext(context.Background(),
  233.         "INSERT INTO `books` (`id`, `title`, `type`, `published_at`, `price`, `stock`) values (?, ?, ?, ?, ?, ?)",
  234.         id, title, bookType, publishedAt, price, stock)
  235.     return err
  236. }
  238. func createUser(txn *util.TiDBSqlTx, id int, nickname string, balance decimal.Decimal) error {
  239.     _, err := txn.ExecContext(context.Background(),
  240.         "INSERT INTO `users` (`id`, `nickname`, `balance`) VALUES (?, ?, ?)",
  241.         id, nickname, balance)
  242.     return err
  243. }

再编写一个包含 main 函数的 txn.go 来调用 helper.go，同时处理传入的命令行参数：

  1. package main
  3. import (
  4.     "database/sql"
  5.     "flag"
  6.     "fmt"
  7.     "sync"
  8. )
  10. func main() {
  11.     optimistic, alice, bob := parseParams()
  13.     openDB("mysql", "root:@tcp(127.0.0.1:4000)/bookshop?charset=utf8mb4", func(db *sql.DB) {
  14.         prepareData(db, optimistic)
  15.         buy(db, optimistic, alice, bob)
  16.     })
  17. }
  19. func buy(db *sql.DB, optimistic bool, alice, bob int) {
  20.     buyFunc := buyOptimistic
  21.     if !optimistic {
  22.         buyFunc = buyPessimistic
  23.     }
  25.     wg := sync.WaitGroup{}
  26.     wg.Add(1)
  27.     go func() {
  28.         defer wg.Done()
  29.         buyFunc(db, 1, 1000, 1, 1, bob)
  30.     }()
  32.     wg.Add(1)
  33.     go func() {
  34.         defer wg.Done()
  35.         buyFunc(db, 2, 1001, 1, 2, alice)
  36.     }()
  38.     wg.Wait()
  39. }
  41. func openDB(driverName, dataSourceName string, runnable func(db *sql.DB)) {
  42.     db, err := sql.Open(driverName, dataSourceName)
  43.     if err != nil {
  44.         panic(err)
  45.     }
  46.     defer db.Close()
  48.     runnable(db)
  49. }
  51. func parseParams() (optimistic bool, alice, bob int) {
  52.     flag.BoolVar(&optimistic, "o", false, "transaction is optimistic")
  53.     flag.IntVar(&alice, "a", 4, "Alice bought num")
  54.     flag.IntVar(&bob, "b", 6, "Bob bought num")
  56.     flag.Parse()
  58.     fmt.Println(optimistic, alice, bob)
  60.     return optimistic, alice, bob
  61. }

Golang 的例子中，已经包含乐观事务。

  1. import time
  3. import MySQLdb
  4. import os
  5. import datetime
  6. from threading import Thread
  8. REPEATABLE_ERROR_CODE_SET = {
  9.     9007,  # Transactions in TiKV encounter write conflicts.
  10.     8028,  # table schema changes
  11.     8002,  # "SELECT FOR UPDATE" commit conflict
  12.     8022   # The transaction commit fails and has been rolled back
  13. }
  16. def create_connection():
  17.     return MySQLdb.connect(
  18.         host="127.0.0.1",
  19.         port=4000,
  20.         user="root",
  21.         password="",
  22.         database="bookshop",
  23.         autocommit=False
  24.     )
  27. def prepare_data() -> None:
  28.     connection = create_connection()
  29.     with connection:
  30.         with connection.cursor() as cursor:
  31.             cursor.execute("INSERT INTO `books` (`id`, `title`, `type`, `published_at`, `price`, `stock`) "
  32.                            "values (%s, %s, %s, %s, %s, %s)",
  33.                            (1, "Designing Data-Intensive Application", "Science & Technology",
  34.                             datetime.datetime(2018, 9, 1), 100, 10))
  36.             cursor.executemany("INSERT INTO `users` (`id`, `nickname`, `balance`) VALUES (%s, %s, %s)",
  37.                                [(1, "Bob", 10000), (2, "ALICE", 10000)])
  38.             connection.commit()
  41. def buy_optimistic(thread_id: int, order_id: int, book_id: int, user_id: int, amount: int,
  42.                    optimistic_retry_times: int = 5) -> None:
  43.     connection = create_connection()
  45.     txn_log_header = f"/* txn {thread_id} */"
  46.     if thread_id != 1:
  47.         txn_log_header = "\t" + txn_log_header
  49.     with connection:
  50.         with connection.cursor() as cursor:
  51.             cursor.execute("BEGIN OPTIMISTIC")
  52.             print(f'{txn_log_header} BEGIN OPTIMISTIC')
  53.             time.sleep(1)
  55.             try:
  56.                 # read the price of book
  57.                 select_book_for_update = "SELECT `price`, `stock` FROM books WHERE id = %s FOR UPDATE"
  58.                 cursor.execute(select_book_for_update, (book_id,))
  59.                 book = cursor.fetchone()
  60.                 if book is None:
  61.                     raise Exception("book_id not exist")
  62.                 price, stock = book
  63.                 print(f'{txn_log_header} {select_book_for_update} successful')
  65.                 if stock < amount:
  66.                     raise Exception("book not enough, rollback")
  68.                 # update book
  69.                 update_stock = "update `books` set stock = stock - %s where id = %s and stock - %s >= 0"
  70.                 rows_affected = cursor.execute(update_stock, (amount, book_id, amount))
  71.                 print(f'{txn_log_header} {update_stock} successful')
  73.                 if rows_affected == 0:
  74.                     raise Exception("stock not enough, rollback")
  76.                 # insert order
  77.                 insert_order = "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (%s, %s, %s, %s)"
  78.                 cursor.execute(insert_order, (order_id, book_id, user_id, amount))
  79.                 print(f'{txn_log_header} {insert_order} successful')
  81.                 # update user
  82.                 update_user = "update `users` set `balance` = `balance` - %s where id = %s"
  83.                 cursor.execute(update_user, (amount * price, user_id))
  84.                 print(f'{txn_log_header} {update_user} successful')
  86.             except Exception as err:
  87.                 connection.rollback()
  89.                 print(f'something went wrong: {err}')
  90.             else:
  91.                 # important here! you need deal the Exception from the TiDB
  92.                 try:
  93.                     connection.commit()
  94.                 except MySQLdb.MySQLError as db_err:
  95.                     code, desc = db_err.args
  96.                     if code in REPEATABLE_ERROR_CODE_SET and optimistic_retry_times > 0:
  97.                         print(f'retry, rest {optimistic_retry_times - 1} times, for {code} {desc}')
  98.                         buy_optimistic(thread_id, order_id, book_id, user_id, amount, optimistic_retry_times - 1)
  101. def buy_pessimistic(thread_id: int, order_id: int, book_id: int, user_id: int, amount: int) -> None:
  102.     connection = create_connection()
  104.     txn_log_header = f"/* txn {thread_id} */"
  105.     if thread_id != 1:
  106.         txn_log_header = "\t" + txn_log_header
  108.     with connection:
  109.         with connection.cursor() as cursor:
  110.             cursor.execute("BEGIN PESSIMISTIC")
  111.             print(f'{txn_log_header} BEGIN PESSIMISTIC')
  112.             time.sleep(1)
  114.             try:
  115.                 # read the price of book
  116.                 select_book_for_update = "SELECT `price` FROM books WHERE id = %s FOR UPDATE"
  117.                 cursor.execute(select_book_for_update, (book_id,))
  118.                 book = cursor.fetchone()
  119.                 if book is None:
  120.                     raise Exception("book_id not exist")
  121.                 price = book[0]
  122.                 print(f'{txn_log_header} {select_book_for_update} successful')
  124.                 # update book
  125.                 update_stock = "update `books` set stock = stock - %s where id = %s and stock - %s >= 0"
  126.                 rows_affected = cursor.execute(update_stock, (amount, book_id, amount))
  127.                 print(f'{txn_log_header} {update_stock} successful')
  129.                 if rows_affected == 0:
  130.                     raise Exception("stock not enough, rollback")
  132.                 # insert order
  133.                 insert_order = "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (%s, %s, %s, %s)"
  134.                 cursor.execute(insert_order, (order_id, book_id, user_id, amount))
  135.                 print(f'{txn_log_header} {insert_order} successful')
  137.                 # update user
  138.                 update_user = "update `users` set `balance` = `balance` - %s where id = %s"
  139.                 cursor.execute(update_user, (amount * price, user_id))
  140.                 print(f'{txn_log_header} {update_user} successful')
  142.             except Exception as err:
  143.                 connection.rollback()
  144.                 print(f'something went wrong: {err}')
  145.             else:
  146.                 connection.commit()
  149. optimistic = os.environ.get('OPTIMISTIC')
  150. alice = os.environ.get('ALICE')
  151. bob = os.environ.get('BOB')
  153. if not (optimistic and alice and bob):
  154.     raise Exception("please use \"OPTIMISTIC=<is_optimistic> ALICE=<alice_num> "
  155.                     "BOB=<bob_num> python3 txn_example.py\" to start this script")
  157. prepare_data()
  159. if bool(optimistic) is True:
  160.     buy_func = buy_optimistic
  161. else:
  162.     buy_func = buy_pessimistic
  164. bob_thread = Thread(target=buy_func, kwargs={
  165.     "thread_id": 1, "order_id": 1000, "book_id": 1, "user_id": 1, "amount": int(bob)})
  166. alice_thread = Thread(target=buy_func, kwargs={
  167.     "thread_id": 2, "order_id": 1001, "book_id": 1, "user_id": 2, "amount": int(alice)})
  169. bob_thread.start()
  170. alice_thread.start()
  171. bob_thread.join(timeout=10)
  172. alice_thread.join(timeout=10)

Python 的例子中，已经包含乐观事务。

2. 运行不涉及超卖的例子

运行示例程序：

  • Java
  • Golang
  • Python

在 Java 中运行示例程序：

  1. mvn clean package
  2. java -jar target/plain-java-txn-0.0.1-jar-with-dependencies.jar ALICE_NUM=4 BOB_NUM=6

在 Golang 中运行示例程序：

  1. go build -o bin/txn
  2. ./bin/txn -a 4 -b 6

在 Python 中运行示例程序：

  1. OPTIMISTIC=False ALICE=4 BOB=6 python3 txn_example.py

SQL 日志：

  1. /* txn 1 */ BEGIN PESSIMISTIC
  2.     /* txn 2 */ BEGIN PESSIMISTIC
  3.     /* txn 2 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
  4.     /* txn 2 */ UPDATE `books` SET `stock` = `stock` - 4 WHERE `id` = 1 AND `stock` - 4 >= 0
  5.     /* txn 2 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1001, 1, 1, 4)
  6.     /* txn 2 */ UPDATE `users` SET `balance` = `balance` - 400.0 WHERE `id` = 2
  7.     /* txn 2 */ COMMIT
  8. /* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
  9. /* txn 1 */ UPDATE `books` SET `stock` = `stock` - 6 WHERE `id` = 1 AND `stock` - 6 >= 0
  10. /* txn 1 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1000, 1, 1, 6)
  11. /* txn 1 */ UPDATE `users` SET `balance` = `balance` - 600.0 WHERE `id` = 1
  12. /* txn 1 */ COMMIT

最后，检验一下订单创建、用户余额扣减、图书库存扣减情况，都符合预期。

  1. mysql> SELECT * FROM `books`;
  2. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  3. | id | title                                | type                 | published_at        | stock | price  |
  4. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  5. |  1 | Designing Data-Intensive Application | Science & Technology | 2018-09-01 00:00:00 |     0 | 100.00 |
  6. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  7. 1 row in set (0.00 sec)
  9. mysql> SELECT * FROM orders;
  10. +------+---------+---------+---------+---------------------+
  11. | id   | book_id | user_id | quality | ordered_at          |
  12. +------+---------+---------+---------+---------------------+
  13. | 1000 |       1 |       1 |       6 | 2022-04-19 10:58:12 |
  14. | 1001 |       1 |       1 |       4 | 2022-04-19 10:58:11 |
  15. +------+---------+---------+---------+---------------------+
  16. 2 rows in set (0.01 sec)
  18. mysql> SELECT * FROM users;
  19. +----+---------+----------+
  20. | id | balance | nickname |
  21. +----+---------+----------+
  22. |  1 | 9400.00 | Bob      |
  23. |  2 | 9600.00 | Alice    |
  24. +----+---------+----------+
  25. 2 rows in set (0.00 sec)

3. 运行防止超卖的例子

可以再把难度加大，如果图书的库存剩余 10 本，Bob 购买 7 本，Alice 购买 4 本，两人几乎同时下单，结果会是怎样？继续复用上个例子里的代码来解决这个需求，只不过把 Bob 购买数量从 6 改成 7：

运行示例程序：

  • Java
  • Golang
  • Python

在 Java 中运行示例程序：

  1. mvn clean package
  2. java -jar target/plain-java-txn-0.0.1-jar-with-dependencies.jar ALICE_NUM=4 BOB_NUM=7

在 Golang 中运行示例程序：

  1. go build -o bin/txn
  2. ./bin/txn -a 4 -b 7

在 Python 中运行示例程序：

  1. OPTIMISTIC=False ALICE=4 BOB=7 python3 txn_example.py
  1. /* txn 1 */ BEGIN PESSIMISTIC
  2.     /* txn 2 */ BEGIN PESSIMISTIC
  3.     /* txn 2 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
  4.     /* txn 2 */ UPDATE `books` SET `stock` = `stock` - 4 WHERE `id` = 1 AND `stock` - 4 >= 0
  5.     /* txn 2 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) values (1001, 1, 1, 4)
  6.     /* txn 2 */ UPDATE `users` SET `balance` = `balance` - 400.0 WHERE `id` = 2
  7.     /* txn 2 */ COMMIT
  8. /* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
  9. /* txn 1 */ UPDATE `books` SET `stock` = `stock` - 7 WHERE `id` = 1 AND `stock` - 7 >= 0
  10. /* txn 1 */ ROLLBACK

由于 txn 2 抢先获得锁资源，更新了 stock，txn 1 里面 affected_rows 返回值为 0，进入了 rollback 流程。

再检验一下订单创建、用户余额扣减、图书库存扣减情况。Alice 下单 4 本书成功，Bob 下单 7 本书失败，库存剩余 6 本符合预期。

  1. mysql> SELECT * FROM books;
  2. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  3. | id | title                                | type                 | published_at        | stock | price  |
  4. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  5. |  1 | Designing Data-Intensive Application | Science & Technology | 2018-09-01 00:00:00 |     6 | 100.00 |
  6. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  7. 1 row in set (0.00 sec)
  9. mysql> SELECT * FROM orders;
  10. +------+---------+---------+---------+---------------------+
  11. | id   | book_id | user_id | quality | ordered_at          |
  12. +------+---------+---------+---------+---------------------+
  13. | 1001 |       1 |       1 |       4 | 2022-04-19 11:03:03 |
  14. +------+---------+---------+---------+---------------------+
  15. 1 row in set (0.00 sec)
  17. mysql> SELECT * FROM users;
  18. +----+----------+----------+
  19. | id | balance  | nickname |
  20. +----+----------+----------+
  21. |  1 | 10000.00 | Bob      |
  22. |  2 |  9600.00 | Alice    |
  23. +----+----------+----------+
  24. 2 rows in set (0.01 sec)

乐观事务

下面代码以乐观事务的方式，用两个线程模拟了两个用户并发买同一本书的过程，和悲观事务的示例一样。书店剩余 10 本，Bob 购买了 6 本，Alice 购买了 4 本。两个人几乎同一时间完成订单，最终，这本书的剩余库存为零。

1. 编写乐观事务示例

  • Java
  • Golang
  • Python

使用 Java 编写乐观事务示例：

代码编写

  1. package com.pingcap.txn.optimistic;
  3. import com.zaxxer.hikari.HikariDataSource;
  5. import java.math.BigDecimal;
  6. import java.sql.*;
  7. import java.util.Arrays;
  8. import java.util.concurrent.*;
  10. public class TxnExample {
  11.     public static void main(String[] args) throws SQLException, InterruptedException {
  12.         System.out.println(Arrays.toString(args));
  13.         int aliceQuantity = 0;
  14.         int bobQuantity = 0;
  16.         for (String arg: args) {
  17.             if (arg.startsWith("ALICE_NUM")) {
  18.                 aliceQuantity = Integer.parseInt(arg.replace("ALICE_NUM=", ""));
  19.             }
  21.             if (arg.startsWith("BOB_NUM")) {
  22.                 bobQuantity = Integer.parseInt(arg.replace("BOB_NUM=", ""));
  23.             }
  24.         }
  26.         HikariDataSource ds = new HikariDataSource();
  27.         ds.setJdbcUrl("jdbc:mysql://localhost:4000/bookshop?useServerPrepStmts=true&cachePrepStmts=true");
  28.         ds.setUsername("root");
  29.         ds.setPassword("");
  31.         // prepare data
  32.         Connection connection = ds.getConnection();
  33.         createBook(connection, 1L, "Designing Data-Intensive Application", "Science & Technology",
  34.                 Timestamp.valueOf("2018-09-01 00:00:00"), new BigDecimal(100), 10);
  35.         createUser(connection, 1L, "Bob", new BigDecimal(10000));
  36.         createUser(connection, 2L, "Alice", new BigDecimal(10000));
  38.         CountDownLatch countDownLatch = new CountDownLatch(2);
  39.         ExecutorService threadPool = Executors.newFixedThreadPool(2);
  41.         final int finalBobQuantity = bobQuantity;
  42.         threadPool.execute(() -> {
  43.             buy(ds, 1, 1000L, 1L, 1L, finalBobQuantity, 5);
  44.             countDownLatch.countDown();
  45.         });
  46.         final int finalAliceQuantity = aliceQuantity;
  47.         threadPool.execute(() -> {
  48.             buy(ds, 2, 1001L, 1L, 2L, finalAliceQuantity, 5);
  49.             countDownLatch.countDown();
  50.         });
  52.         countDownLatch.await(5, TimeUnit.SECONDS);
  53.     }
  55.     public static void createUser(Connection connection, Long id, String nickname, BigDecimal balance) throws SQLException  {
  56.         PreparedStatement insert = connection.prepareStatement(
  57.                 "INSERT INTO `users` (`id`, `nickname`, `balance`) VALUES (?, ?, ?)");
  58.         insert.setLong(1, id);
  59.         insert.setString(2, nickname);
  60.         insert.setBigDecimal(3, balance);
  61.         insert.executeUpdate();
  62.     }
  64.     public static void createBook(Connection connection, Long id, String title, String type, Timestamp publishedAt, BigDecimal price, Integer stock) throws SQLException {
  65.         PreparedStatement insert = connection.prepareStatement(
  66.                 "INSERT INTO `books` (`id`, `title`, `type`, `published_at`, `price`, `stock`) values (?, ?, ?, ?, ?, ?)");
  67.         insert.setLong(1, id);
  68.         insert.setString(2, title);
  69.         insert.setString(3, type);
  70.         insert.setTimestamp(4, publishedAt);
  71.         insert.setBigDecimal(5, price);
  72.         insert.setInt(6, stock);
  74.         insert.executeUpdate();
  75.     }
  77.     public static void buy (HikariDataSource ds, Integer threadID, Long orderID, Long bookID,
  78.                             Long userID, Integer quantity, Integer retryTimes) {
  79.         String txnComment = "/* txn " + threadID + " */ ";
  81.         try (Connection connection = ds.getConnection()) {
  82.             try {
  84.                 connection.setAutoCommit(false);
  85.                 connection.createStatement().executeUpdate(txnComment + "begin optimistic");
  87.                 // waiting for other thread ran the 'begin optimistic' statement
  88.                 TimeUnit.SECONDS.sleep(1);
  90.                 BigDecimal price = null;
  92.                 // read price of book
  93.                 PreparedStatement selectBook = connection.prepareStatement(txnComment + "SELECT * FROM books where id = ? for update");
  94.                 selectBook.setLong(1, bookID);
  95.                 ResultSet res = selectBook.executeQuery();
  96.                 if (!res.next()) {
  97.                     throw new RuntimeException("book not exist");
  98.                 } else {
  99.                     price = res.getBigDecimal("price");
  100.                     int stock = res.getInt("stock");
  101.                     if (stock < quantity) {
  102.                         throw new RuntimeException("book not enough");
  103.                     }
  104.                 }
  106.                 // update book
  107.                 String updateBookSQL = "update `books` set stock = stock - ? where id = ? and stock - ? >= 0";
  108.                 PreparedStatement updateBook = connection.prepareStatement(txnComment + updateBookSQL);
  109.                 updateBook.setInt(1, quantity);
  110.                 updateBook.setLong(2, bookID);
  111.                 updateBook.setInt(3, quantity);
  112.                 updateBook.executeUpdate();
  114.                 // insert order
  115.                 String insertOrderSQL = "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (?, ?, ?, ?)";
  116.                 PreparedStatement insertOrder = connection.prepareStatement(txnComment + insertOrderSQL);
  117.                 insertOrder.setLong(1, orderID);
  118.                 insertOrder.setLong(2, bookID);
  119.                 insertOrder.setLong(3, userID);
  120.                 insertOrder.setInt(4, quantity);
  121.                 insertOrder.executeUpdate();
  123.                 // update user
  124.                 String updateUserSQL = "update `users` set `balance` = `balance` - ? where id = ?";
  125.                 PreparedStatement updateUser = connection.prepareStatement(txnComment + updateUserSQL);
  126.                 updateUser.setBigDecimal(1, price.multiply(new BigDecimal(quantity)));
  127.                 updateUser.setLong(2, userID);
  128.                 updateUser.executeUpdate();
  130.                 connection.createStatement().executeUpdate(txnComment + "commit");
  131.             } catch (Exception e) {
  132.                 connection.createStatement().executeUpdate(txnComment + "rollback");
  133.                 System.out.println("error occurred: " + e.getMessage());
  135.                 if (e instanceof SQLException sqlException) {
  136.                     switch (sqlException.getErrorCode()) {
  137.                         // You can get all error codes at https://docs.pingcap.com/tidb/stable/error-codes
  138.                         case 9007: // Transactions in TiKV encounter write conflicts.
  139.                         case 8028: // table schema changes
  140.                         case 8002: // "SELECT FOR UPDATE" commit conflict
  141.                         case 8022: // The transaction commit fails and has been rolled back
  142.                             if (retryTimes != 0) {
  143.                                 System.out.println("rest " + retryTimes + " times. retry for " + e.getMessage());
  144.                                 buy(ds, threadID, orderID, bookID, userID, quantity, retryTimes - 1);
  145.                             }
  146.                     }
  147.                 }
  148.             }
  149.         } catch (SQLException e) {
  150.             e.printStackTrace();
  151.         }
  152.     }
  153. }

配置更改

此处，需将 pom.xml 中启动类：

  1. <mainClass>com.pingcap.txn.TxnExample</mainClass>

更改为：

  1. <mainClass>com.pingcap.txn.optimistic.TxnExample</mainClass>

来指向乐观事务的例子。

Golang 在编写悲观事务示例章节中的例子已经支持了乐观事务，无需更改，可直接使用。

Python 在编写悲观事务示例章节中的例子已经支持了乐观事务，无需更改，可直接使用。

2. 运行不涉及超卖的例子

运行示例程序：

  • Java
  • Golang
  • Python

在 Java 中运行示例程序：

  1. mvn clean package
  2. java -jar target/plain-java-txn-0.0.1-jar-with-dependencies.jar ALICE_NUM=4 BOB_NUM=6

在 Golang 中运行示例程序：

  1. go build -o bin/txn
  2. ./bin/txn -a 4 -b 6 -o true

在 Python 中运行示例程序：

  1. OPTIMISTIC=True ALICE=4 BOB=6 python3 txn_example.py

SQL 语句执行过程：

  1.     /* txn 2 */ BEGIN OPTIMISTIC
  2. /* txn 1 */ BEGIN OPTIMISTIC
  3.     /* txn 2 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
  4.     /* txn 2 */ UPDATE `books` SET `stock` = `stock` - 4 WHERE `id` = 1 AND `stock` - 4 >= 0
  5.     /* txn 2 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1001, 1, 1, 4)
  6.     /* txn 2 */ UPDATE `users` SET `balance` = `balance` - 400.0 WHERE `id` = 2
  7.     /* txn 2 */ COMMIT
  8. /* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 for UPDATE
  9. /* txn 1 */ UPDATE `books` SET `stock` = `stock` - 6 WHERE `id` = 1 AND `stock` - 6 >= 0
  10. /* txn 1 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1000, 1, 1, 6)
  11. /* txn 1 */ UPDATE `users` SET `balance` = `balance` - 600.0 WHERE `id` = 1
  12. retry 1 times for 9007 Write conflict, txnStartTS=432618733006225412, conflictStartTS=432618733006225411, conflictCommitTS=432618733006225414, key={tableID=126, handle=1} primary={tableID=114, indexID=1, indexValues={1, 1000, }} [try again later]
  13. /* txn 1 */ BEGIN OPTIMISTIC
  14. /* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
  15. /* txn 1 */ UPDATE `books` SET `stock` = `stock` - 6 WHERE `id` = 1 AND `stock` - 6 >= 0
  16. /* txn 1 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1000, 1, 1, 6)
  17. /* txn 1 */ UPDATE `users` SET `balance` = `balance` - 600.0 WHERE `id` = 1
  18. /* txn 1 */ COMMIT

在乐观事务模式下，由于中间状态不一定正确，不能像悲观事务模式一样，通过 affected_rows 来判断某个语句是否执行成功。需要把事务看做一个整体，通过最终的 COMMIT 语句是否返回异常来判断当前事务是否发生写冲突。

从上面 SQL 日志可以看出，由于两个事务并发执行，并且对同一条记录做了修改，txn 1 COMMIT 之后抛出了 9007 Write conflict 异常。对于乐观事务写冲突，在应用端可以进行安全的重试，重试一次之后提交成功，最终执行结果符合预期：

  1. mysql> SELECT * FROM books;
  2. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  3. | id | title                                | type                 | published_at        | stock | price  |
  4. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  5. |  1 | Designing Data-Intensive Application | Science & Technology | 2018-09-01 00:00:00 |     0 | 100.00 |
  6. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  7. 1 row in set (0.01 sec)
  9. mysql> SELECT * FROM orders;
  10. +------+---------+---------+---------+---------------------+
  11. | id   | book_id | user_id | quality | ordered_at          |
  12. +------+---------+---------+---------+---------------------+
  13. | 1000 |       1 |       1 |       6 | 2022-04-19 03:18:19 |
  14. | 1001 |       1 |       1 |       4 | 2022-04-19 03:18:17 |
  15. +------+---------+---------+---------+---------------------+
  16. 2 rows in set (0.01 sec)
  18. mysql> SELECT * FROM users;
  19. +----+---------+----------+
  20. | id | balance | nickname |
  21. +----+---------+----------+
  22. |  1 | 9400.00 | Bob      |
  23. |  2 | 9600.00 | Alice    |
  24. +----+---------+----------+
  25. 2 rows in set (0.00 sec)

3. 运行防止超卖的例子

再来看一下用乐观事务防止超卖的例子，如果图书的库存剩余 10 本，Bob 购买 7 本，Alice 购买 4 本，两人几乎同时下单，结果会是怎样？继续复用乐观事务例子里的代码来解决这个需求，只不过把 Bob 购买数量从 6 改成 7：

运行示例程序：

  • Java
  • Golang
  • Python

在 Java 中运行示例程序：

  1. mvn clean package
  2. java -jar target/plain-java-txn-0.0.1-jar-with-dependencies.jar ALICE_NUM=4 BOB_NUM=7

在 Golang 中运行示例程序：

  1. go build -o bin/txn
  2. ./bin/txn -a 4 -b 7 -o true

在 Python 中运行示例程序：

  1. OPTIMISTIC=True ALICE=4 BOB=7 python3 txn_example.py
  1. /* txn 1 */ BEGIN OPTIMISTIC
  2.     /* txn 2 */ BEGIN OPTIMISTIC
  3.     /* txn 2 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
  4.     /* txn 2 */ UPDATE `books` SET `stock` = `stock` - 4 WHERE `id` = 1 AND `stock` - 4 >= 0
  5.     /* txn 2 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1001, 1, 1, 4)
  6.     /* txn 2 */ UPDATE `users` SET `balance` = `balance` - 400.0 WHERE `id` = 2
  7.     /* txn 2 */ COMMIT
  8. /* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
  9. /* txn 1 */ UPDATE `books` SET `stock` = `stock` - 7 WHERE `id` = 1 AND `stock` - 7 >= 0
  10. /* txn 1 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1000, 1, 1, 7)
  11. /* txn 1 */ UPDATE `users` SET `balance` = `balance` - 700.0 WHERE `id` = 1
  12. retry 1 times for 9007 Write conflict, txnStartTS=432619094333980675, conflictStartTS=432619094333980676, conflictCommitTS=432619094333980678, key={tableID=126, handle=1} primary={tableID=114, indexID=1, indexValues={1, 1000, }} [try again later]
  13. /* txn 1 */ BEGIN OPTIMISTIC
  14. /* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
  15. Fail -> out of stock
  16. /* txn 1 */ ROLLBACK

从上面的 SQL 日志可以看出，第一次执行由于写冲突，txn 1 在应用端进行了重试，从获取到的最新快照对比发现，剩余库存不够，应用端抛出 out of stock 异常结束。

  1. mysql> SELECT * FROM books;
  2. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  3. | id | title                                | type                 | published_at        | stock | price  |
  4. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  5. |  1 | Designing Data-Intensive Application | Science & Technology | 2018-09-01 00:00:00 |     6 | 100.00 |
  6. +----+--------------------------------------+----------------------+---------------------+-------+--------+
  7. 1 row in set (0.00 sec)
  9. mysql> SELECT * FROM orders;
  10. +------+---------+---------+---------+---------------------+
  11. | id   | book_id | user_id | quality | ordered_at          |
  12. +------+---------+---------+---------+---------------------+
  13. | 1001 |       1 |       1 |       4 | 2022-04-19 03:41:16 |
  14. +------+---------+---------+---------+---------------------+
  15. 1 row in set (0.00 sec)
  17. mysql> SELECT * FROM users;
  18. +----+----------+----------+
  19. | id | balance  | nickname |
  20. +----+----------+----------+
  21. |  1 | 10000.00 | Bob      |
  22. |  2 |  9600.00 | Alice    |
  23. +----+----------+----------+
  24. 2 rows in set (0.00 sec)