## 1. 案例背景
– 系统：TiDB 7.5.0
– 业务：电商平台商品库存管理
– 需求：使用乐观锁处理并发库存扣减

## 2. 实施步骤

### 步骤1：创建测试表
– 创建商品表：
[root@fgedu.net.cn ~]# mysql -h 192.168.1.1 -P 4000 -u fgedu -p fgedudb
mysql> CREATE TABLE fgedu_products (
mysql> id INT PRIMARY KEY AUTO_INCREMENT,
mysql> name VARCHAR(100) NOT NULL,
mysql> price DECIMAL(10,2) NOT NULL,
mysql> stock INT NOT NULL,
mysql> version INT DEFAULT 1
mysql> ) ENGINE=InnoDB;

### 步骤2：插入测试数据
– 插入商品数据：
mysql> INSERT INTO fgedu_products (name, price, stock) VALUES (‘product1’, 100.00, 100);

### 步骤3：实现乐观锁更新
– 会话1：
mysql> START TRANSACTION;
mysql> SELECT id, stock, version FROM fgedu_products WHERE id = 1;

结果：
+—-+——-+———+
| id | stock | version |
+—-+——-+———+
| 1 | 100 | 1 |
+—-+——-+———+

mysql> UPDATE fgedu_products SET stock = stock – 1, version = version + 1 WHERE id = 1 AND version = 1;

结果：
Query OK, 1 row affected (0.01 sec)
Rows matched: 1 Changed: 1 Warnings: 0

mysql> COMMIT;

– 会话2：
mysql> START TRANSACTION;
mysql> SELECT id, stock, version FROM fgedu_products WHERE id = 1;

结果：
+—-+——-+———+
| id | stock | version |
+—-+——-+———+
| 1 | 100 | 1 |
+—-+——-+———+

mysql> UPDATE fgedu_products SET stock = stock – 1, version = version + 1 WHERE id = 1 AND version = 1;

结果：
Query OK, 0 rows affected (0.01 sec)
Rows matched: 0 Changed: 0 Warnings: 0

mysql> SELECT id, stock, version FROM fgedu_products WHERE id = 1;

结果：
+—-+——-+———+
| id | stock | version |
+—-+——-+———+
| 1 | 99 | 2 |
+—-+——-+———+

mysql> UPDATE fgedu_products SET stock = stock – 1, version = version + 1 WHERE id = 1 AND version = 2;

结果：
Query OK, 1 row affected (0.01 sec)
Rows matched: 1 Changed: 1 Warnings: 0

mysql> COMMIT;

### 步骤4：验证结果
– 查看商品库存：
mysql> SELECT id, stock, version FROM fgedu_products WHERE id = 1;

结果：
+—-+——-+———+
| id | stock | version |
+—-+——-+———+
| 1 | 98 | 3 |
+—-+——-+———+

## 3. 案例效果
– 乐观锁实现成功：并发更新时通过版本号控制
– 冲突处理：会话2发现冲突后重新读取并更新
– 数据一致性：最终库存正确扣减
– 并发性能：无需加锁，提高并发性能

## 1. 案例背景
– 系统：TiDB 7.5.0
– 业务：秒杀活动
– 问题：高并发下库存扣减冲突严重

## 2. 实施步骤

### 步骤1：创建测试表
– 创建商品表：
[root@fgedu.net.cn ~]# mysql -h 192.168.1.1 -P 4000 -u fgedu -p fgedudb
mysql> CREATE TABLE fgedu_seckill_products (
mysql> id INT PRIMARY KEY AUTO_INCREMENT,
mysql> name VARCHAR(100) NOT NULL,
mysql> price DECIMAL(10,2) NOT NULL,
mysql> stock INT NOT NULL,
mysql> version INT DEFAULT 1
mysql> ) ENGINE=InnoDB;

### 步骤2：插入测试数据
– 插入秒杀商品：
mysql> INSERT INTO fgedu_seckill_products (name, price, stock) VALUES (‘seckill_product’, 1.00, 10);

### 步骤3：实现冲突处理
– 编写Python脚本：
import pymysql
import time
import random

def update_stock(product_id, quantity):
max_retries = 5
base_delay = 0.1

for i in range(max_retries):
try:
conn = pymysql.connect(
host=’192.168.1.1′,
port=4000,
user=’fgedu’,
password=’password’,
db=’fgedudb’
)
cursor = conn.cursor()

# 开始事务
conn.begin()

# 读取数据
cursor.execute(“SELECT stock, version FROM fgedu_seckill_products WHERE id = %s”, (product_id,))
result = cursor.fetchone()
if not result:
conn.rollback()
cursor.close()
conn.close()
return False

stock, version = result
if stock < quantity: conn.rollback() cursor.close() conn.close() return False # 尝试更新 cursor.execute( "UPDATE fgedu_seckill_products SET stock = stock - %s, version = version + 1 WHERE id = %s AND version = %s", (quantity, product_id, version) ) if cursor.rowcount == 0: # 冲突，重试 conn.rollback() cursor.close() conn.close() delay = min(base_delay * (2 ** i) + random.uniform(0, 0.1), 1.0) time.sleep(delay) continue # 提交事务 conn.commit() cursor.close() conn.close() return True except Exception as e: if conn: conn.rollback() cursor.close() conn.close() if 'Write conflict' in str(e): delay = min(base_delay * (2 ** i) + random.uniform(0, 0.1), 1.0) time.sleep(delay) continue else: raise return False ### 步骤4：模拟并发测试 - 编写并发测试脚本： import threading import time def test_update_stock(): success = update_stock(1, 1) print(f"Update {'success' if success else 'failed'}") # 创建15个线程模拟并发 threads = [] for i in range(15): t = threading.Thread(target=test_update_stock) threads.append(t) t.start() # 等待所有线程完成 for t in threads: t.join() ### 步骤5：验证结果 - 查看商品库存： mysql> SELECT id, stock, version FROM fgedu_seckill_products WHERE id = 1;

结果：
+—-+——-+———+
| id | stock | version |
+—-+——-+———+
| 1 | 0 | 10 |
+—-+——-+———+

## 3. 案例效果
– 并发测试成功：10个线程成功扣减库存
– 冲突处理：5个线程失败（库存不足）
– 数据一致性：最终库存为0，正确
– 性能表现：并发处理速度快，无死锁

## 1. 案例背景
– 系统：TiDB 7.5.0
– 业务：银行转账系统
– 需求：实现可靠的转账操作，处理并发冲突

## 2. 实施步骤

### 步骤1：创建测试表
– 创建账户表：
[root@fgedu.net.cn ~]# mysql -h 192.168.1.1 -P 4000 -u fgedu -p fgedudb
mysql> CREATE TABLE fgedu_accounts (
mysql> id INT PRIMARY KEY AUTO_INCREMENT,
mysql> name VARCHAR(50) NOT NULL,
mysql> balance DECIMAL(10,2) DEFAULT 0,
mysql> version INT DEFAULT 1
mysql> ) ENGINE=InnoDB;

### 步骤2：插入测试数据
– 插入账户数据：
mysql> INSERT INTO fgedu_accounts (name, balance) VALUES (‘account1’, 1000.00), (‘account2′, 1000.00);

### 步骤3：实现转账功能
– 编写Python脚本：
import pymysql
import time

def transfer(from_account, to_account, amount):
max_retries = 3
base_delay = 0.1

for i in range(max_retries):
try:
conn = pymysql.connect(
host=’192.168.1.1′,
port=4000,
user=’fgedu’,
password=’password’,
db=’fgedudb’
)
cursor = conn.cursor()

# 开始事务
conn.begin()

# 读取转出账户
cursor.execute(“SELECT balance, version FROM fgedu_accounts WHERE id = %s”, (from_account,))
from_result = cursor.fetchone()
if not from_result:
conn.rollback()
cursor.close()
conn.close()
return False, “From account not found”

from_balance, from_version = from_result
if from_balance < amount: conn.rollback() cursor.close() conn.close() return False, "Insufficient balance" # 读取转入账户 cursor.execute("SELECT balance, version FROM fgedu_accounts WHERE id = %s", (to_account,)) to_result = cursor.fetchone() if not to_result: conn.rollback() cursor.close() conn.close() return False, "To account not found" to_balance, to_version = to_result # 更新转出账户 cursor.execute( "UPDATE fgedu_accounts SET balance = balance - %s, version = version + 1 WHERE id = %s AND version = %s", (amount, from_account, from_version) ) if cursor.rowcount == 0: # 冲突，重试 conn.rollback() cursor.close() conn.close() delay = base_delay * (2 ** i) time.sleep(delay) continue # 更新转入账户 cursor.execute( "UPDATE fgedu_accounts SET balance = balance + %s, version = version + 1 WHERE id = %s AND version = %s", (amount, to_account, to_version) ) if cursor.rowcount == 0: # 冲突，重试 conn.rollback() cursor.close() conn.close() delay = base_delay * (2 ** i) time.sleep(delay) continue # 提交事务 conn.commit() cursor.close() conn.close() return True, "Transfer successful" except Exception as e: if conn: conn.rollback() cursor.close() conn.close() if 'Write conflict' in str(e): delay = base_delay * (2 ** i) time.sleep(delay) continue else: return False, str(e) return False, "Max retries exceeded" ### 步骤4：模拟并发测试 - 编写并发测试脚本： import threading def test_transfer(): success, message = transfer(1, 2, 100) print(f"Transfer {'success' if success else 'failed'}: {message}") # 创建5个线程模拟并发转账 threads = [] for i in range(5): t = threading.Thread(target=test_transfer) threads.append(t) t.start() # 等待所有线程完成 for t in threads: t.join() ### 步骤5：验证结果 - 查看账户余额： mysql> SELECT id, name, balance, version FROM fgedu_accounts;

结果：
+—-+———-+———+———+
| id | name | balance | version |
+—-+———-+———+———+
| 1 | account1 | 500.00 | 6 |
| 2 | account2 | 1500.00 | 6 |
+—-+———-+———+———+

## 3. 案例效果
– 并发测试成功：5次转账操作全部成功
– 重试机制：处理了并发冲突
– 数据一致性：账户余额正确
– 可靠性：转账操作原子性保证