```java
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();
}
}
}
```
此处，需将 `pom.xml` 中启动类
```xml
com.pingcap.txn.TxnExample
```
更改为：
```xml
com.pingcap.txn.optimistic.TxnExample
```
来指向乐观事务的例子
运行示例程序：
```shell
mvn clean package
java -jar target/plain-java-txn-0.0.1-jar-with-dependencies.jar ALICE_NUM=4 BOB_NUM=7
```
```sql
/* 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 */ rollback
```
从上面的 SQL 日志可以看出，第一次执行由于写冲突，`txn 1` 在应用端进行了重试，从获取到的最新快照对比发现，剩余库存不够，应用端抛出 `out of stock` 异常结束。
```sql
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)
```