| title | summary |
|---|---|
Optimistic Transactions and Pessimistic Transactions |
Learn about optimistic and pessimistic transactions in TiDB. |
The optimistic transaction model commits the transaction directly, and rolls back when there is a conflict. By contrast, the pessimistic transaction model tries to lock the resources that need to be modified before actually committing the transaction, and only starts committing after ensuring that the transaction can be successfully executed.
The optimistic transaction model is suitable for scenarios with low conflict rates, because the direct commit has a high probability of success. But once a transaction conflict occurs, the cost of rollback is relatively high.
The advantage of the pessimistic transaction model is that for scenarios with high conflict rates, the cost of locking ahead is less than the cost of rollback afterwards. Moreover, it can solve the problem that multiple concurrent transactions fail to commit due to conflicts. However, the pessimistic transaction model is not as efficient as the optimistic transaction model in scenarios with low conflict rates.
The pessimistic transaction model is more intuitive and easier to implement on the application side. The optimistic transaction model requires complex application-side retry mechanisms.
The following is an example of a bookshop. It uses an example of buying books to show the pros and cons of optimistic and pessimistic transactions. The process of buying books mainly consists of the following:
- Update the stock quantity
- Create an order
- Make the payment
These operations must either all succeed or all fail. You must ensure that overselling does not happen in the case of concurrent transactions.
The following code uses two threads to simulate the process that two users buy the same book in a pessimistic transaction mode. There are 10 books left in the bookstore. Bob buys 6 books, and Alice buys 4 books. They complete the orders at nearly the same time. As a result, all books in inventory are sold out.
Because you use multiple threads to simulate the situation that multiple users insert data simultaneously, you need to use a connection object with safe threads. Here use Java's popular connection pool HikariCP for demo.
sql.DB in Golang is concurrency-safe, so there is no need to import a third-party package.
To adapt TiDB transactions, write a toolkit util according to the following code:
package util
import (
"context"
"database/sql"
)
type TiDBSqlTx struct {
*sql.Tx
conn *sql.Conn
pessimistic bool
}
func TiDBSqlBegin(db *sql.DB, pessimistic bool) (*TiDBSqlTx, error) {
ctx := context.Background()
conn, err := db.Conn(ctx)
if err != nil {
return nil, err
}
if pessimistic {
_, err = conn.ExecContext(ctx, "set @@tidb_txn_mode=?", "pessimistic")
} else {
_, err = conn.ExecContext(ctx, "set @@tidb_txn_mode=?", "optimistic")
}
if err != nil {
return nil, err
}
tx, err := conn.BeginTx(ctx, nil)
if err != nil {
return nil, err
}
return &TiDBSqlTx{
conn: conn,
Tx: tx,
pessimistic: pessimistic,
}, nil
}
func (tx *TiDBSqlTx) Commit() error {
defer tx.conn.Close()
return tx.Tx.Commit()
}
func (tx *TiDBSqlTx) Rollback() error {
defer tx.conn.Close()
return tx.Tx.Rollback()
}To ensure thread safety, you can use the mysqlclient driver to open multiple connections that are not shared between threads.
Configuration file
If you use Maven to manage the package, in the <dependencies> node in pom.xml, add the following dependencies to import HikariCP, and set the packaging target, and the main class of the JAR package startup. The following is an example of pom.xml.
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.pingcap</groupId>
<artifactId>plain-java-txn</artifactId>
<version>0.0.1</version>
<name>plain-java-jdbc</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13.2</version>
<scope>test</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.28</version>
</dependency>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
<version>5.0.1</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.3.0</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass>com.pingcap.txn.TxnExample</mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>Coding
Then write the code:
package com.pingcap.txn;
import com.zaxxer.hikari.HikariDataSource;
import java.math.BigDecimal;
import java.sql.*;
import java.util.Arrays;
import java.util.concurrent.*;
public class TxnExample {
public static void main(String[] args) throws SQLException, InterruptedException {
System.out.println(Arrays.toString(args));
int aliceQuantity = 0;
int bobQuantity = 0;
for (String arg: args) {
if (arg.startsWith("ALICE_NUM")) {
aliceQuantity = Integer.parseInt(arg.replace("ALICE_NUM=", ""));
}
if (arg.startsWith("BOB_NUM")) {
bobQuantity = Integer.parseInt(arg.replace("BOB_NUM=", ""));
}
}
HikariDataSource ds = new HikariDataSource();
ds.setJdbcUrl("jdbc:mysql://localhost:4000/bookshop?useServerPrepStmts=true&cachePrepStmts=true");
ds.setUsername("root");
ds.setPassword("");
// prepare data
Connection connection = ds.getConnection();
createBook(connection, 1L, "Designing Data-Intensive Application", "Science & Technology",
Timestamp.valueOf("2018-09-01 00:00:00"), new BigDecimal(100), 10);
createUser(connection, 1L, "Bob", new BigDecimal(10000));
createUser(connection, 2L, "Alice", new BigDecimal(10000));
CountDownLatch countDownLatch = new CountDownLatch(2);
ExecutorService threadPool = Executors.newFixedThreadPool(2);
final int finalBobQuantity = bobQuantity;
threadPool.execute(() -> {
buy(ds, 1, 1000L, 1L, 1L, finalBobQuantity);
countDownLatch.countDown();
});
final int finalAliceQuantity = aliceQuantity;
threadPool.execute(() -> {
buy(ds, 2, 1001L, 1L, 2L, finalAliceQuantity);
countDownLatch.countDown();
});
countDownLatch.await(5, TimeUnit.SECONDS);
}
public static void createUser(Connection connection, Long id, String nickname, BigDecimal balance) throws SQLException {
PreparedStatement insert = connection.prepareStatement(
"INSERT INTO `users` (`id`, `nickname`, `balance`) VALUES (?, ?, ?)");
insert.setLong(1, id);
insert.setString(2, nickname);
insert.setBigDecimal(3, balance);
insert.executeUpdate();
}
public static void createBook(Connection connection, Long id, String title, String type, Timestamp publishedAt, BigDecimal price, Integer stock) throws SQLException {
PreparedStatement insert = connection.prepareStatement(
"INSERT INTO `books` (`id`, `title`, `type`, `published_at`, `price`, `stock`) values (?, ?, ?, ?, ?, ?)");
insert.setLong(1, id);
insert.setString(2, title);
insert.setString(3, type);
insert.setTimestamp(4, publishedAt);
insert.setBigDecimal(5, price);
insert.setInt(6, stock);
insert.executeUpdate();
}
public static void buy (HikariDataSource ds, Integer threadID,
Long orderID, Long bookID, Long userID, Integer quantity) {
String txnComment = "/* txn " + threadID + " */ ";
try (Connection connection = ds.getConnection()) {
try {
connection.setAutoCommit(false);
connection.createStatement().executeUpdate(txnComment + "begin pessimistic");
// waiting for other thread ran the 'begin pessimistic' statement
TimeUnit.SECONDS.sleep(1);
BigDecimal price = null;
// read price of book
PreparedStatement selectBook = connection.prepareStatement(txnComment + "select price from books where id = ? for update");
selectBook.setLong(1, bookID);
ResultSet res = selectBook.executeQuery();
if (!res.next()) {
throw new RuntimeException("book not exist");
} else {
price = res.getBigDecimal("price");
}
// update book
String updateBookSQL = "update `books` set stock = stock - ? where id = ? and stock - ? >= 0";
PreparedStatement updateBook = connection.prepareStatement(txnComment + updateBookSQL);
updateBook.setInt(1, quantity);
updateBook.setLong(2, bookID);
updateBook.setInt(3, quantity);
int affectedRows = updateBook.executeUpdate();
if (affectedRows == 0) {
// stock not enough, rollback
connection.createStatement().executeUpdate(txnComment + "rollback");
return;
}
// insert order
String insertOrderSQL = "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (?, ?, ?, ?)";
PreparedStatement insertOrder = connection.prepareStatement(txnComment + insertOrderSQL);
insertOrder.setLong(1, orderID);
insertOrder.setLong(2, bookID);
insertOrder.setLong(3, userID);
insertOrder.setInt(4, quantity);
insertOrder.executeUpdate();
// update user
String updateUserSQL = "update `users` set `balance` = `balance` - ? where id = ?";
PreparedStatement updateUser = connection.prepareStatement(txnComment + updateUserSQL);
updateUser.setBigDecimal(1, price.multiply(new BigDecimal(quantity)));
updateUser.setLong(2, userID);
updateUser.executeUpdate();
connection.createStatement().executeUpdate(txnComment + "commit");
} catch (Exception e) {
connection.createStatement().executeUpdate(txnComment + "rollback");
e.printStackTrace();
}
} catch (SQLException e) {
e.printStackTrace();
}
}
}Write a helper.go file that contains the required database operations:
package main
import (
"context"
"database/sql"
"fmt"
"time"
"github.com/go-sql-driver/mysql"
"github.com/pingcap-inc/tidb-example-golang/util"
"github.com/shopspring/decimal"
)
type TxnFunc func(txn *util.TiDBSqlTx) error
const (
ErrWriteConflict = 9007 // Transactions in TiKV encounter write conflicts.
ErrInfoSchemaChanged = 8028 // table schema changes
ErrForUpdateCantRetry = 8002 // "SELECT FOR UPDATE" commit conflict
ErrTxnRetryable = 8022 // The transaction commit fails and has been rolled back
)
const retryTimes = 5
var retryErrorCodeSet = map[uint16]interface{}{
ErrWriteConflict: nil,
ErrInfoSchemaChanged: nil,
ErrForUpdateCantRetry: nil,
ErrTxnRetryable: nil,
}
func runTxn(db *sql.DB, optimistic bool, optimisticRetryTimes int, txnFunc TxnFunc) {
txn, err := util.TiDBSqlBegin(db, !optimistic)
if err != nil {
panic(err)
}
err = txnFunc(txn)
if err != nil {
txn.Rollback()
if mysqlErr, ok := err.(*mysql.MySQLError); ok && optimistic && optimisticRetryTimes != 0 {
if _, retryableError := retryErrorCodeSet[mysqlErr.Number]; retryableError {
fmt.Printf("[runTxn] got a retryable error, rest time: %d\n", optimisticRetryTimes-1)
runTxn(db, optimistic, optimisticRetryTimes-1, txnFunc)
return
}
}
fmt.Printf("[runTxn] got an error, rollback: %+v\n", err)
} else {
err = txn.Commit()
if mysqlErr, ok := err.(*mysql.MySQLError); ok && optimistic && optimisticRetryTimes != 0 {
if _, retryableError := retryErrorCodeSet[mysqlErr.Number]; retryableError {
fmt.Printf("[runTxn] got a retryable error, rest time: %d\n", optimisticRetryTimes-1)
runTxn(db, optimistic, optimisticRetryTimes-1, txnFunc)
return
}
}
if err == nil {
fmt.Println("[runTxn] commit success")
}
}
}
func prepareData(db *sql.DB, optimistic bool) {
runTxn(db, optimistic, retryTimes, func(txn *util.TiDBSqlTx) error {
publishedAt, err := time.Parse("2006-01-02 15:04:05", "2018-09-01 00:00:00")
if err != nil {
return err
}
if err = createBook(txn, 1, "Designing Data-Intensive Application",
"Science & Technology", publishedAt, decimal.NewFromInt(100), 10); err != nil {
return err
}
if err = createUser(txn, 1, "Bob", decimal.NewFromInt(10000)); err != nil {
return err
}
if err = createUser(txn, 2, "Alice", decimal.NewFromInt(10000)); err != nil {
return err
}
return nil
})
}
func buyPessimistic(db *sql.DB, goroutineID, orderID, bookID, userID, amount int) {
txnComment := fmt.Sprintf("/* txn %d */ ", goroutineID)
if goroutineID != 1 {
txnComment = "\t" + txnComment
}
fmt.Printf("\nuser %d try to buy %d books(id: %d)\n", userID, amount, bookID)
runTxn(db, false, retryTimes, func(txn *util.TiDBSqlTx) error {
time.Sleep(time.Second)
// read the price of book
selectBookForUpdate := "select `price` from books where id = ? for update"
bookRows, err := txn.Query(selectBookForUpdate, bookID)
if err != nil {
return err
}
fmt.Println(txnComment + selectBookForUpdate + " successful")
defer bookRows.Close()
price := decimal.NewFromInt(0)
if bookRows.Next() {
err = bookRows.Scan(&price)
if err != nil {
return err
}
} else {
return fmt.Errorf("book ID not exist")
}
bookRows.Close()
// update book
updateStock := "update `books` set stock = stock - ? where id = ? and stock - ? >= 0"
result, err := txn.Exec(updateStock, amount, bookID, amount)
if err != nil {
return err
}
fmt.Println(txnComment + updateStock + " successful")
affected, err := result.RowsAffected()
if err != nil {
return err
}
if affected == 0 {
return fmt.Errorf("stock not enough, rollback")
}
// insert order
insertOrder := "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (?, ?, ?, ?)"
if _, err := txn.Exec(insertOrder,
orderID, bookID, userID, amount); err != nil {
return err
}
fmt.Println(txnComment + insertOrder + " successful")
// update user
updateUser := "update `users` set `balance` = `balance` - ? where id = ?"
if _, err := txn.Exec(updateUser,
price.Mul(decimal.NewFromInt(int64(amount))), userID); err != nil {
return err
}
fmt.Println(txnComment + updateUser + " successful")
return nil
})
}
func buyOptimistic(db *sql.DB, goroutineID, orderID, bookID, userID, amount int) {
txnComment := fmt.Sprintf("/* txn %d */ ", goroutineID)
if goroutineID != 1 {
txnComment = "\t" + txnComment
}
fmt.Printf("\nuser %d try to buy %d books(id: %d)\n", userID, amount, bookID)
runTxn(db, true, retryTimes, func(txn *util.TiDBSqlTx) error {
time.Sleep(time.Second)
// read the price and stock of book
selectBookForUpdate := "select `price`, `stock` from books where id = ? for update"
bookRows, err := txn.Query(selectBookForUpdate, bookID)
if err != nil {
return err
}
fmt.Println(txnComment + selectBookForUpdate + " successful")
defer bookRows.Close()
price, stock := decimal.NewFromInt(0), 0
if bookRows.Next() {
err = bookRows.Scan(&price, &stock)
if err != nil {
return err
}
} else {
return fmt.Errorf("book ID not exist")
}
bookRows.Close()
if stock < amount {
return fmt.Errorf("book not enough")
}
// update book
updateStock := "update `books` set stock = stock - ? where id = ? and stock - ? >= 0"
result, err := txn.Exec(updateStock, amount, bookID, amount)
if err != nil {
return err
}
fmt.Println(txnComment + updateStock + " successful")
affected, err := result.RowsAffected()
if err != nil {
return err
}
if affected == 0 {
return fmt.Errorf("stock not enough, rollback")
}
// insert order
insertOrder := "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (?, ?, ?, ?)"
if _, err := txn.Exec(insertOrder,
orderID, bookID, userID, amount); err != nil {
return err
}
fmt.Println(txnComment + insertOrder + " successful")
// update user
updateUser := "update `users` set `balance` = `balance` - ? where id = ?"
if _, err := txn.Exec(updateUser,
price.Mul(decimal.NewFromInt(int64(amount))), userID); err != nil {
return err
}
fmt.Println(txnComment + updateUser + " successful")
return nil
})
}
func createBook(txn *util.TiDBSqlTx, id int, title, bookType string,
publishedAt time.Time, price decimal.Decimal, stock int) error {
_, err := txn.ExecContext(context.Background(),
"INSERT INTO `books` (`id`, `title`, `type`, `published_at`, `price`, `stock`) values (?, ?, ?, ?, ?, ?)",
id, title, bookType, publishedAt, price, stock)
return err
}
func createUser(txn *util.TiDBSqlTx, id int, nickname string, balance decimal.Decimal) error {
_, err := txn.ExecContext(context.Background(),
"INSERT INTO `users` (`id`, `nickname`, `balance`) VALUES (?, ?, ?)",
id, nickname, balance)
return err
}Then write a txn.go with a main function to call helper.go and handle the incoming command line arguments:
package main
import (
"database/sql"
"flag"
"fmt"
"sync"
)
func main() {
optimistic, alice, bob := parseParams()
openDB("mysql", "root:@tcp(127.0.0.1:4000)/bookshop?charset=utf8mb4", func(db *sql.DB) {
prepareData(db, optimistic)
buy(db, optimistic, alice, bob)
})
}
func buy(db *sql.DB, optimistic bool, alice, bob int) {
buyFunc := buyOptimistic
if !optimistic {
buyFunc = buyPessimistic
}
wg := sync.WaitGroup{}
wg.Add(1)
go func() {
defer wg.Done()
buyFunc(db, 1, 1000, 1, 1, bob)
}()
wg.Add(1)
go func() {
defer wg.Done()
buyFunc(db, 2, 1001, 1, 2, alice)
}()
wg.Wait()
}
func openDB(driverName, dataSourceName string, runnable func(db *sql.DB)) {
db, err := sql.Open(driverName, dataSourceName)
if err != nil {
panic(err)
}
defer db.Close()
runnable(db)
}
func parseParams() (optimistic bool, alice, bob int) {
flag.BoolVar(&optimistic, "o", false, "transaction is optimistic")
flag.IntVar(&alice, "a", 4, "Alice bought num")
flag.IntVar(&bob, "b", 6, "Bob bought num")
flag.Parse()
fmt.Println(optimistic, alice, bob)
return optimistic, alice, bob
}The Golang example already includes optimistic transactions.
import time
import MySQLdb
import os
import datetime
from threading import Thread
REPEATABLE_ERROR_CODE_SET = {
9007, # Transactions in TiKV encounter write conflicts.
8028, # table schema changes
8002, # "SELECT FOR UPDATE" commit conflict
8022 # The transaction commit fails and has been rolled back
}
def create_connection():
return MySQLdb.connect(
host="127.0.0.1",
port=4000,
user="root",
password="",
database="bookshop",
autocommit=False
)
def prepare_data() -> None:
connection = create_connection()
with connection:
with connection.cursor() as cursor:
cursor.execute("INSERT INTO `books` (`id`, `title`, `type`, `published_at`, `price`, `stock`) "
"values (%s, %s, %s, %s, %s, %s)",
(1, "Designing Data-Intensive Application", "Science & Technology",
datetime.datetime(2018, 9, 1), 100, 10))
cursor.executemany("INSERT INTO `users` (`id`, `nickname`, `balance`) VALUES (%s, %s, %s)",
[(1, "Bob", 10000), (2, "ALICE", 10000)])
connection.commit()
def buy_optimistic(thread_id: int, order_id: int, book_id: int, user_id: int, amount: int,
optimistic_retry_times: int = 5) -> None:
connection = create_connection()
txn_log_header = f"/* txn {thread_id} */"
if thread_id != 1:
txn_log_header = "\t" + txn_log_header
with connection:
with connection.cursor() as cursor:
cursor.execute("BEGIN OPTIMISTIC")
print(f'{txn_log_header} BEGIN OPTIMISTIC')
time.sleep(1)
try:
# read the price of book
select_book_for_update = "SELECT `price`, `stock` FROM books WHERE id = %s FOR UPDATE"
cursor.execute(select_book_for_update, (book_id,))
book = cursor.fetchone()
if book is None:
raise Exception("book_id not exist")
price, stock = book
print(f'{txn_log_header} {select_book_for_update} successful')
if stock < amount:
raise Exception("book not enough, rollback")
# update book
update_stock = "update `books` set stock = stock - %s where id = %s and stock - %s >= 0"
rows_affected = cursor.execute(update_stock, (amount, book_id, amount))
print(f'{txn_log_header} {update_stock} successful')
if rows_affected == 0:
raise Exception("stock not enough, rollback")
# insert order
insert_order = "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (%s, %s, %s, %s)"
cursor.execute(insert_order, (order_id, book_id, user_id, amount))
print(f'{txn_log_header} {insert_order} successful')
# update user
update_user = "update `users` set `balance` = `balance` - %s where id = %s"
cursor.execute(update_user, (amount * price, user_id))
print(f'{txn_log_header} {update_user} successful')
except Exception as err:
connection.rollback()
print(f'something went wrong: {err}')
else:
# important here! you need deal the Exception from the TiDB
try:
connection.commit()
except MySQLdb.MySQLError as db_err:
code, desc = db_err.args
if code in REPEATABLE_ERROR_CODE_SET and optimistic_retry_times > 0:
print(f'retry, rest {optimistic_retry_times - 1} times, for {code} {desc}')
buy_optimistic(thread_id, order_id, book_id, user_id, amount, optimistic_retry_times - 1)
def buy_pessimistic(thread_id: int, order_id: int, book_id: int, user_id: int, amount: int) -> None:
connection = create_connection()
txn_log_header = f"/* txn {thread_id} */"
if thread_id != 1:
txn_log_header = "\t" + txn_log_header
with connection:
with connection.cursor() as cursor:
cursor.execute("BEGIN PESSIMISTIC")
print(f'{txn_log_header} BEGIN PESSIMISTIC')
time.sleep(1)
try:
# read the price of book
select_book_for_update = "SELECT `price` FROM books WHERE id = %s FOR UPDATE"
cursor.execute(select_book_for_update, (book_id,))
book = cursor.fetchone()
if book is None:
raise Exception("book_id not exist")
price = book[0]
print(f'{txn_log_header} {select_book_for_update} successful')
# update book
update_stock = "update `books` set stock = stock - %s where id = %s and stock - %s >= 0"
rows_affected = cursor.execute(update_stock, (amount, book_id, amount))
print(f'{txn_log_header} {update_stock} successful')
if rows_affected == 0:
raise Exception("stock not enough, rollback")
# insert order
insert_order = "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (%s, %s, %s, %s)"
cursor.execute(insert_order, (order_id, book_id, user_id, amount))
print(f'{txn_log_header} {insert_order} successful')
# update user
update_user = "update `users` set `balance` = `balance` - %s where id = %s"
cursor.execute(update_user, (amount * price, user_id))
print(f'{txn_log_header} {update_user} successful')
except Exception as err:
connection.rollback()
print(f'something went wrong: {err}')
else:
connection.commit()
optimistic = os.environ.get('OPTIMISTIC')
alice = os.environ.get('ALICE')
bob = os.environ.get('BOB')
if not (optimistic and alice and bob):
raise Exception("please use \"OPTIMISTIC=<is_optimistic> ALICE=<alice_num> "
"BOB=<bob_num> python3 txn_example.py\" to start this script")
prepare_data()
if bool(optimistic) is True:
buy_func = buy_optimistic
else:
buy_func = buy_pessimistic
bob_thread = Thread(target=buy_func, kwargs={
"thread_id": 1, "order_id": 1000, "book_id": 1, "user_id": 1, "amount": int(bob)})
alice_thread = Thread(target=buy_func, kwargs={
"thread_id": 2, "order_id": 1001, "book_id": 1, "user_id": 2, "amount": int(alice)})
bob_thread.start()
alice_thread.start()
bob_thread.join(timeout=10)
alice_thread.join(timeout=10)The Python example already includes optimistic transactions.
Run the sample program:
mvn clean package
java -jar target/plain-java-txn-0.0.1-jar-with-dependencies.jar ALICE_NUM=4 BOB_NUM=6go build -o bin/txn
./bin/txn -a 4 -b 6OPTIMISTIC=False ALICE=4 BOB=6 python3 txn_example.pySQL logs:
/* txn 1 */ BEGIN PESSIMISTIC
/* txn 2 */ BEGIN PESSIMISTIC
/* txn 2 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
/* txn 2 */ UPDATE `books` SET `stock` = `stock` - 4 WHERE `id` = 1 AND `stock` - 4 >= 0
/* txn 2 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1001, 1, 1, 4)
/* txn 2 */ UPDATE `users` SET `balance` = `balance` - 400.0 WHERE `id` = 2
/* txn 2 */ COMMIT
/* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
/* txn 1 */ UPDATE `books` SET `stock` = `stock` - 6 WHERE `id` = 1 AND `stock` - 6 >= 0
/* txn 1 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1000, 1, 1, 6)
/* txn 1 */ UPDATE `users` SET `balance` = `balance` - 600.0 WHERE `id` = 1
/* txn 1 */ COMMITFinally, check that the order is created, the user balance is deducted, and the book inventory is deducted as expected.
mysql> SELECT * FROM `books`;
+----+--------------------------------------+----------------------+---------------------+-------+--------+
| id | title | type | published_at | stock | price |
+----+--------------------------------------+----------------------+---------------------+-------+--------+
| 1 | Designing Data-Intensive Application | Science & Technology | 2018-09-01 00:00:00 | 0 | 100.00 |
+----+--------------------------------------+----------------------+---------------------+-------+--------+
1 row in set (0.00 sec)
mysql> SELECT * FROM orders;
+------+---------+---------+---------+---------------------+
| id | book_id | user_id | quality | ordered_at |
+------+---------+---------+---------+---------------------+
| 1000 | 1 | 1 | 6 | 2022-04-19 10:58:12 |
| 1001 | 1 | 1 | 4 | 2022-04-19 10:58:11 |
+------+---------+---------+---------+---------------------+
2 rows in set (0.01 sec)
mysql> SELECT * FROM users;
+----+---------+----------+
| id | balance | nickname |
+----+---------+----------+
| 1 | 9400.00 | Bob |
| 2 | 9600.00 | Alice |
+----+---------+----------+
2 rows in set (0.00 sec)The task in this example is more challenging. Suppose there are 10 books left in stock. Bob buys 7 books, Alice buys 4 books, and they place orders almost at the same time. What will happen? You can reuse the code from the previous example to solve this challenge, and change Bob's purchase quantity from 6 to 7.
Run the sample program:
mvn clean package
java -jar target/plain-java-txn-0.0.1-jar-with-dependencies.jar ALICE_NUM=4 BOB_NUM=7go build -o bin/txn
./bin/txn -a 4 -b 7OPTIMISTIC=False ALICE=4 BOB=7 python3 txn_example.py/* txn 1 */ BEGIN PESSIMISTIC
/* txn 2 */ BEGIN PESSIMISTIC
/* txn 2 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
/* txn 2 */ UPDATE `books` SET `stock` = `stock` - 4 WHERE `id` = 1 AND `stock` - 4 >= 0
/* txn 2 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) values (1001, 1, 1, 4)
/* txn 2 */ UPDATE `users` SET `balance` = `balance` - 400.0 WHERE `id` = 2
/* txn 2 */ COMMIT
/* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
/* txn 1 */ UPDATE `books` SET `stock` = `stock` - 7 WHERE `id` = 1 AND `stock` - 7 >= 0
/* txn 1 */ ROLLBACKSince txn 2 preemptively gets the lock resource and updates the stock, the return value of affected_rows in txn 1 is 0, and it enters the rollback process.
Let's check the order creation, user balance deduction, and book inventory deduction. Alice successfully ordered 4 books, Bob failed to order 7 books, and the remaining 6 books are in stock as expected.
mysql> SELECT * FROM books;
+----+--------------------------------------+----------------------+---------------------+-------+--------+
| id | title | type | published_at | stock | price |
+----+--------------------------------------+----------------------+---------------------+-------+--------+
| 1 | Designing Data-Intensive Application | Science & Technology | 2018-09-01 00:00:00 | 6 | 100.00 |
+----+--------------------------------------+----------------------+---------------------+-------+--------+
1 row in set (0.00 sec)
mysql> SELECT * FROM orders;
+------+---------+---------+---------+---------------------+
| id | book_id | user_id | quality | ordered_at |
+------+---------+---------+---------+---------------------+
| 1001 | 1 | 1 | 4 | 2022-04-19 11:03:03 |
+------+---------+---------+---------+---------------------+
1 row in set (0.00 sec)
mysql> SELECT * FROM users;
+----+----------+----------+
| id | balance | nickname |
+----+----------+----------+
| 1 | 10000.00 | Bob |
| 2 | 9600.00 | Alice |
+----+----------+----------+
2 rows in set (0.01 sec)The following code uses two threads to simulate the process that two users buy the same book in an optimistic transaction, just like the pessimistic transaction example. There are 10 books left in inventory. Bob buys 6 and Alice buys 4. They complete the order at about the same time. In the end, no books are left in inventory.
Coding
package com.pingcap.txn.optimistic;
import com.zaxxer.hikari.HikariDataSource;
import java.math.BigDecimal;
import java.sql.*;
import java.util.Arrays;
import java.util.concurrent.*;
public class TxnExample {
public static void main(String[] args) throws SQLException, InterruptedException {
System.out.println(Arrays.toString(args));
int aliceQuantity = 0;
int bobQuantity = 0;
for (String arg: args) {
if (arg.startsWith("ALICE_NUM")) {
aliceQuantity = Integer.parseInt(arg.replace("ALICE_NUM=", ""));
}
if (arg.startsWith("BOB_NUM")) {
bobQuantity = Integer.parseInt(arg.replace("BOB_NUM=", ""));
}
}
HikariDataSource ds = new HikariDataSource();
ds.setJdbcUrl("jdbc:mysql://localhost:4000/bookshop?useServerPrepStmts=true&cachePrepStmts=true");
ds.setUsername("root");
ds.setPassword("");
// prepare data
Connection connection = ds.getConnection();
createBook(connection, 1L, "Designing Data-Intensive Application", "Science & Technology",
Timestamp.valueOf("2018-09-01 00:00:00"), new BigDecimal(100), 10);
createUser(connection, 1L, "Bob", new BigDecimal(10000));
createUser(connection, 2L, "Alice", new BigDecimal(10000));
CountDownLatch countDownLatch = new CountDownLatch(2);
ExecutorService threadPool = Executors.newFixedThreadPool(2);
final int finalBobQuantity = bobQuantity;
threadPool.execute(() -> {
buy(ds, 1, 1000L, 1L, 1L, finalBobQuantity, 5);
countDownLatch.countDown();
});
final int finalAliceQuantity = aliceQuantity;
threadPool.execute(() -> {
buy(ds, 2, 1001L, 1L, 2L, finalAliceQuantity, 5);
countDownLatch.countDown();
});
countDownLatch.await(5, TimeUnit.SECONDS);
}
public static void createUser(Connection connection, Long id, String nickname, BigDecimal balance) throws SQLException {
PreparedStatement insert = connection.prepareStatement(
"INSERT INTO `users` (`id`, `nickname`, `balance`) VALUES (?, ?, ?)");
insert.setLong(1, id);
insert.setString(2, nickname);
insert.setBigDecimal(3, balance);
insert.executeUpdate();
}
public static void createBook(Connection connection, Long id, String title, String type, Timestamp publishedAt, BigDecimal price, Integer stock) throws SQLException {
PreparedStatement insert = connection.prepareStatement(
"INSERT INTO `books` (`id`, `title`, `type`, `published_at`, `price`, `stock`) values (?, ?, ?, ?, ?, ?)");
insert.setLong(1, id);
insert.setString(2, title);
insert.setString(3, type);
insert.setTimestamp(4, publishedAt);
insert.setBigDecimal(5, price);
insert.setInt(6, stock);
insert.executeUpdate();
}
public static void buy (HikariDataSource ds, Integer threadID, Long orderID, Long bookID,
Long userID, Integer quantity, Integer retryTimes) {
String txnComment = "/* txn " + threadID + " */ ";
try (Connection connection = ds.getConnection()) {
try {
connection.setAutoCommit(false);
connection.createStatement().executeUpdate(txnComment + "begin optimistic");
// waiting for other thread ran the 'begin optimistic' statement
TimeUnit.SECONDS.sleep(1);
BigDecimal price = null;
// read price of book
PreparedStatement selectBook = connection.prepareStatement(txnComment + "SELECT * FROM books where id = ? for update");
selectBook.setLong(1, bookID);
ResultSet res = selectBook.executeQuery();
if (!res.next()) {
throw new RuntimeException("book not exist");
} else {
price = res.getBigDecimal("price");
int stock = res.getInt("stock");
if (stock < quantity) {
throw new RuntimeException("book not enough");
}
}
// update book
String updateBookSQL = "update `books` set stock = stock - ? where id = ? and stock - ? >= 0";
PreparedStatement updateBook = connection.prepareStatement(txnComment + updateBookSQL);
updateBook.setInt(1, quantity);
updateBook.setLong(2, bookID);
updateBook.setInt(3, quantity);
updateBook.executeUpdate();
// insert order
String insertOrderSQL = "insert into `orders` (`id`, `book_id`, `user_id`, `quality`) values (?, ?, ?, ?)";
PreparedStatement insertOrder = connection.prepareStatement(txnComment + insertOrderSQL);
insertOrder.setLong(1, orderID);
insertOrder.setLong(2, bookID);
insertOrder.setLong(3, userID);
insertOrder.setInt(4, quantity);
insertOrder.executeUpdate();
// update user
String updateUserSQL = "update `users` set `balance` = `balance` - ? where id = ?";
PreparedStatement updateUser = connection.prepareStatement(txnComment + updateUserSQL);
updateUser.setBigDecimal(1, price.multiply(new BigDecimal(quantity)));
updateUser.setLong(2, userID);
updateUser.executeUpdate();
connection.createStatement().executeUpdate(txnComment + "commit");
} catch (Exception e) {
connection.createStatement().executeUpdate(txnComment + "rollback");
System.out.println("error occurred: " + e.getMessage());
if (e instanceof SQLException sqlException) {
switch (sqlException.getErrorCode()) {
// You can get all error codes at https://docs.pingcap.com/tidb/stable/error-codes
case 9007: // Transactions in TiKV encounter write conflicts.
case 8028: // table schema changes
case 8002: // "SELECT FOR UPDATE" commit conflict
case 8022: // The transaction commit fails and has been rolled back
if (retryTimes != 0) {
System.out.println("rest " + retryTimes + " times. retry for " + e.getMessage());
buy(ds, threadID, orderID, bookID, userID, quantity, retryTimes - 1);
}
}
}
}
} catch (SQLException e) {
e.printStackTrace();
}
}
}Configuration changes
Change the startup class in pom.xml:
<mainClass>com.pingcap.txn.TxnExample</mainClass>Change it to the following to point to the optimistic transaction example.
<mainClass>com.pingcap.txn.optimistic.TxnExample</mainClass>The Golang example in the Write a pessimistic transaction example section already supports optimistic transactions and can be used directly without changes.
The Python example in the Write a pessimistic transaction example section already supports optimistic transactions and can be used directly without changes.
Run the sample program:
mvn clean package
java -jar target/plain-java-txn-0.0.1-jar-with-dependencies.jar ALICE_NUM=4 BOB_NUM=6go build -o bin/txn
./bin/txn -a 4 -b 6 -o trueOPTIMISTIC=True ALICE=4 BOB=6 python3 txn_example.pySQL statement execution process:
/* txn 2 */ BEGIN OPTIMISTIC
/* txn 1 */ BEGIN OPTIMISTIC
/* txn 2 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
/* txn 2 */ UPDATE `books` SET `stock` = `stock` - 4 WHERE `id` = 1 AND `stock` - 4 >= 0
/* txn 2 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1001, 1, 1, 4)
/* txn 2 */ UPDATE `users` SET `balance` = `balance` - 400.0 WHERE `id` = 2
/* txn 2 */ COMMIT
/* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 for UPDATE
/* txn 1 */ UPDATE `books` SET `stock` = `stock` - 6 WHERE `id` = 1 AND `stock` - 6 >= 0
/* txn 1 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1000, 1, 1, 6)
/* txn 1 */ UPDATE `users` SET `balance` = `balance` - 600.0 WHERE `id` = 1
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]
/* txn 1 */ BEGIN OPTIMISTIC
/* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
/* txn 1 */ UPDATE `books` SET `stock` = `stock` - 6 WHERE `id` = 1 AND `stock` - 6 >= 0
/* txn 1 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1000, 1, 1, 6)
/* txn 1 */ UPDATE `users` SET `balance` = `balance` - 600.0 WHERE `id` = 1
/* txn 1 */ COMMITIn the optimistic transaction mode, because the intermediate state is not necessarily correct, it is not possible to judge whether a statement is successfully executed through affected_rows as in the pessimistic transaction mode. You need to regard the transaction as a whole, and judge whether the current transaction has a write conflict by checking whether the final COMMIT statement returns an exception.
As you can see from the above SQL log, because two transactions are executed concurrently and the same record is modified, a 9007 Write conflict exception is thrown after txn 1 COMMIT. For write conflicts in the optimistic transaction mode, you can safely retry on the application side. After one retry, the data is committed successfully. The final execution result is as expected:
mysql> SELECT * FROM books;
+----+--------------------------------------+----------------------+---------------------+-------+--------+
| id | title | type | published_at | stock | price |
+----+--------------------------------------+----------------------+---------------------+-------+--------+
| 1 | Designing Data-Intensive Application | Science & Technology | 2018-09-01 00:00:00 | 0 | 100.00 |
+----+--------------------------------------+----------------------+---------------------+-------+--------+
1 row in set (0.01 sec)
mysql> SELECT * FROM orders;
+------+---------+---------+---------+---------------------+
| id | book_id | user_id | quality | ordered_at |
+------+---------+---------+---------+---------------------+
| 1000 | 1 | 1 | 6 | 2022-04-19 03:18:19 |
| 1001 | 1 | 1 | 4 | 2022-04-19 03:18:17 |
+------+---------+---------+---------+---------------------+
2 rows in set (0.01 sec)
mysql> SELECT * FROM users;
+----+---------+----------+
| id | balance | nickname |
+----+---------+----------+
| 1 | 9400.00 | Bob |
| 2 | 9600.00 | Alice |
+----+---------+----------+
2 rows in set (0.00 sec)This section describes an optimistic transaction example that prevents overselling. Suppose there are 10 books left in inventory. Bob buys 7 books,and Alice buys 4 books. They place orders almost at the same time. What will happen? You can reuse the code from the optimistic transaction example to address this requirement. Change Bob's purchases from 6 to 7.
Run the sample program:
mvn clean package
java -jar target/plain-java-txn-0.0.1-jar-with-dependencies.jar ALICE_NUM=4 BOB_NUM=7go build -o bin/txn
./bin/txn -a 4 -b 7 -o trueOPTIMISTIC=True ALICE=4 BOB=7 python3 txn_example.py/* txn 1 */ BEGIN OPTIMISTIC
/* txn 2 */ BEGIN OPTIMISTIC
/* txn 2 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
/* txn 2 */ UPDATE `books` SET `stock` = `stock` - 4 WHERE `id` = 1 AND `stock` - 4 >= 0
/* txn 2 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1001, 1, 1, 4)
/* txn 2 */ UPDATE `users` SET `balance` = `balance` - 400.0 WHERE `id` = 2
/* txn 2 */ COMMIT
/* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
/* txn 1 */ UPDATE `books` SET `stock` = `stock` - 7 WHERE `id` = 1 AND `stock` - 7 >= 0
/* txn 1 */ INSERT INTO `orders` (`id`, `book_id`, `user_id`, `quality`) VALUES (1000, 1, 1, 7)
/* txn 1 */ UPDATE `users` SET `balance` = `balance` - 700.0 WHERE `id` = 1
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]
/* txn 1 */ BEGIN OPTIMISTIC
/* txn 1 */ SELECT * FROM `books` WHERE `id` = 1 FOR UPDATE
Fail -> out of stock
/* txn 1 */ ROLLBACKYou can see from the above SQL log that txn 1 is retried on the application side due to a write conflict in the first execution. By comparing the latest snapshots, you can find that the stock is running out. The application side throws out of stock, and ends abnormally.
mysql> SELECT * FROM books;
+----+--------------------------------------+----------------------+---------------------+-------+--------+
| id | title | type | published_at | stock | price |
+----+--------------------------------------+----------------------+---------------------+-------+--------+
| 1 | Designing Data-Intensive Application | Science & Technology | 2018-09-01 00:00:00 | 6 | 100.00 |
+----+--------------------------------------+----------------------+---------------------+-------+--------+
1 row in set (0.00 sec)
mysql> SELECT * FROM orders;
+------+---------+---------+---------+---------------------+
| id | book_id | user_id | quality | ordered_at |
+------+---------+---------+---------+---------------------+
| 1001 | 1 | 1 | 4 | 2022-04-19 03:41:16 |
+------+---------+---------+---------+---------------------+
1 row in set (0.00 sec)
mysql> SELECT * FROM users;
+----+----------+----------+
| id | balance | nickname |
+----+----------+----------+
| 1 | 10000.00 | Bob |
| 2 | 9600.00 | Alice |
+----+----------+----------+
2 rows in set (0.00 sec)Ask questions on TiDB Community, or create a support ticket.
Ask questions on TiDB Community, or create a support ticket.