乐观事务和悲观事务

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

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

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

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

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

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

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

    • Java
    • Golang
    • Python

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

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

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

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

    • 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.             if err != nil {
    113.                 return err
    114.             }
    115.         } else {
    116.             return fmt.Errorf("book ID not exist")
    117.         }
    120.         // update book
    121.         updateStock := "update `books` set stock = stock - ? where id = ? and stock - ? >= 0"
    122.         result, err := txn.Exec(updateStock, amount, bookID, amount)
    123.         if err != nil {
    124.             return err
    125.         }
    126.         fmt.Println(txnComment + updateStock + " successful")
    128.         affected, err := result.RowsAffected()
    129.         if err != nil {
    130.             return err
    131.         }
    133.         if affected == 0 {
    134.             return fmt.Errorf("stock not enough, rollback")
    135.         }
    137.         // insert order
    138.         insertOrder := "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (?, ?, ?, ?)"
    139.         if _, err := txn.Exec(insertOrder,
    140.             orderID, bookID, userID, amount); err != nil {
    141.             return err
    142.         }
    143.         fmt.Println(txnComment + insertOrder + " successful")
    145.         // update user
    146.         updateUser := "update `users` set `balance` = `balance` - ? where id = ?"
    147.         if _, err := txn.Exec(updateUser,
    148.             price.Mul(decimal.NewFromInt(int64(amount))), userID); err != nil {
    149.             return err
    150.         }
    151.         fmt.Println(txnComment + updateUser + " successful")
    153.         return nil
    154.     })
    155. }
    157. func buyOptimistic(db *sql.DB, goroutineID, orderID, bookID, userID, amount int) {
    158.     txnComment := fmt.Sprintf("/* txn %d */ ", goroutineID)
    159.     if goroutineID != 1 {
    160.         txnComment = "\t" + txnComment
    161.     }
    163.     fmt.Printf("\nuser %d try to buy %d books(id: %d)\n", userID, amount, bookID)
    165.     runTxn(db, true, retryTimes, func(txn *util.TiDBSqlTx) error {
    166.         time.Sleep(time.Second)
    168.         // read the price and stock of book
    169.         selectBookForUpdate := "select `price`, `stock` from books where id = ? for update"
    170.         bookRows, err := txn.Query(selectBookForUpdate, bookID)
    171.         if err != nil {
    172.             return err
    173.         }
    174.         fmt.Println(txnComment + selectBookForUpdate + " successful")
    175.         defer bookRows.Close()
    177.         price, stock := decimal.NewFromInt(0), 0
    178.         if bookRows.Next() {
    179.             err = bookRows.Scan(&price, &stock)
    180.             if err != nil {
    181.                 return err
    182.             }
    183.         } else {
    184.             return fmt.Errorf("book ID not exist")
    185.         }
    186.         bookRows.Close()
    188.         if stock < amount {
    189.             return fmt.Errorf("book not enough")
    190.         }
    192.         // update book
    193.         updateStock := "update `books` set stock = stock - ? where id = ? and stock - ? >= 0"
    194.         result, err := txn.Exec(updateStock, amount, bookID, amount)
    195.         if err != nil {
    196.             return err
    197.         }
    198.         fmt.Println(txnComment + updateStock + " successful")
    200.         affected, err := result.RowsAffected()
    201.         if err != nil {
    202.             return err
    203.         }
    205.         if affected == 0 {
    206.             return fmt.Errorf("stock not enough, rollback")
    207.         }
    209.         // insert order
    210.         insertOrder := "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (?, ?, ?, ?)"
    211.         if _, err := txn.Exec(insertOrder,
    212.             orderID, bookID, userID, amount); err != nil {
    213.             return err
    214.         }
    215.         fmt.Println(txnComment + insertOrder + " successful")
    217.         // update user
    218.         updateUser := "update `users` set `balance` = `balance` - ? where id = ?"
    219.         if _, err := txn.Exec(updateUser,
    220.             price.Mul(decimal.NewFromInt(int64(amount))), userID); err != nil {
    221.             return err
    222.         }
    223.         fmt.Println(txnComment + updateUser + " successful")
    225.         return nil
    226.     })
    227. }
    229. func createBook(txn *util.TiDBSqlTx, id int, title, bookType string,
    230.     publishedAt time.Time, price decimal.Decimal, stock int) error {
    231.     _, err := txn.ExecContext(context.Background(),
    232.         "INSERT INTO `books` (`id`, `title`, `type`, `published_at`, `price`, `stock`) values (?, ?, ?, ?, ?, ?)",
    233.         id, title, bookType, publishedAt, price, stock)
    234.     return err
    235. }
    237. func createUser(txn *util.TiDBSqlTx, id int, nickname string, balance decimal.Decimal) error {
    238.     _, err := txn.ExecContext(context.Background(),
    239.         "INSERT INTO `users` (`id`, `nickname`, `balance`) VALUES (?, ?, ?)",
    240.         id, nickname, balance)
    241.     return err
    242. }

    再编写一个包含 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()
    148. optimistic = os.environ.get('OPTIMISTIC')
    149. alice = os.environ.get('ALICE')
    150. bob = os.environ.get('BOB')
    152. if not (optimistic and alice and bob):
    153.     raise Exception("please use \"OPTIMISTIC=<is_optimistic> ALICE=<alice_num> "
    154.                     "BOB=<bob_num> python3 txn_example.py\" to start this script")
    156. prepare_data()
    158. if bool(optimistic) is True:
    159.     buy_func = buy_optimistic
    160. else:
    161.     buy_func = buy_pessimistic
    163. bob_thread = Thread(target=buy_func, kwargs={
    164.     "thread_id": 1, "order_id": 1000, "book_id": 1, "user_id": 1, "amount": int(bob)})
    165. alice_thread = Thread(target=buy_func, kwargs={
    166.     "thread_id": 2, "order_id": 1001, "book_id": 1, "user_id": 2, "amount": int(alice)})
    168. bob_thread.start()
    169. alice_thread.start()
    170. bob_thread.join(timeout=10)
    171. alice_thread.join(timeout=10)

    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. 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)

    可以再把难度加大，如果图书的库存剩余 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. }

    配置更改

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

    更改为：

    来指向乐观事务的例子。

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

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

    运行示例程序：

    • 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

    从上面的 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. +----+----------+----------+