乐观锁完全指南（Python 实现版）：原理与实战

文档目标：使用 Python 实现乐观锁的核心机制，基于 MySQL 和 sqlalchemy 演示版本号控制的完整流程。
适用人群：Python 后端开发、数据工程师
技术栈：Python 3.8+, MySQL 8.0, SQLAlchemy, Flask

---

一、引言：并发更新问题

在电商系统中，多个用户同时下单可能导致库存超卖：

# 用户A：查询库存
stock_a = db.query("SELECT stock, version FROM products WHERE id=1")
# 返回: stock=10, version=0

# 用户B：也查询库存
stock_b = db.query("SELECT stock, version FROM products WHERE id=1")
# 返回: stock=10, version=0

# 用户A：扣减库存
db.execute("UPDATE products SET stock=9, version=1 WHERE id=1 AND version=0")

# 用户B：也扣减库存（但 version 已变为 1）
db.execute("UPDATE products SET stock=9, version=1 WHERE id=1 AND version=0")  # ❌ 失败！

👉 乐观锁通过 version 字段确保只有第一个请求成功，避免超卖。

---

二、乐观锁实现原理

核心思想

• ✅ 读取时不加锁
• 🔁 更新时检查版本号是否匹配
• ❌ 不匹配则拒绝更新

SQL 更新语句

UPDATE products 
SET stock = %s, version = version + 1 
WHERE id = %s AND version = %s;

• 如果返回 rowcount == 1 → 成功
• 如果返回 0 → 版本冲突，需重试

---

三、完整实现（Python + SQLAlchemy）

1. 环境准备

安装依赖

pip install sqlalchemy flask pymysql retrying

创建数据库表

CREATE DATABASE optimistic_lock_demo CHARACTER SET utf8mb4;

USE optimistic_lock_demo;

CREATE TABLE product (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(100) NOT NULL,
    stock INT NOT NULL DEFAULT 0,
    price DECIMAL(10,2),
    version INT DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 插入测试数据
INSERT INTO product (name, stock, price) VALUES ('iPhone 15', 10, 5999.00);

---

2. 数据库连接配置

# database.py
from sqlalchemy import create_engine, Column, Integer, String, DECIMAL, TIMESTAMP, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os

# 数据库配置
DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/optimistic_lock_demo"

# 创建引擎
engine = create_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,  # 连接前检测
    echo=False  # 开发环境可设为 True 查看 SQL
)

# 基类
Base = declarative_base()

# 创建会话
SessionLocal = sessionmaker(bind=engine)

---

3. ORM 模型定义

# models.py
from .database import Base
from sqlalchemy import Column, Integer, String, DECIMAL, TIMESTAMP, text

class Product(Base):
    __tablename__ = 'product'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    stock = Column(Integer, nullable=False, default=0)
    price = Column(DECIMAL(10, 2))
    version = Column(Integer, nullable=False, default=0)
    created_at = Column(TIMESTAMP, server_default=text('CURRENT_TIMESTAMP'))
    updated_at = Column(TIMESTAMP, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP'))

    def __repr__(self):
        return f"<Product(id={self.id}, name='{self.name}', stock={self.stock}, version={self.version})>"

---

4. 服务层（带重试逻辑）

# services.py
from sqlalchemy.exc import SQLAlchemyError
from .models import Product
from .database import SessionLocal
from retrying import retry
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProductService:
    def __init__(self):
        self.session = SessionLocal()

    def get_product(self, product_id: int) -> Product:
        """查询商品"""
        product = self.session.query(Product).filter(Product.id == product_id).first()
        if not product:
            raise ValueError(f"商品不存在: {product_id}")
        return product

    @retry(
        stop_max_attempt_number=5,  # 最大重试5次
        wait_exponential_multiplier=100,  # 指数退避：100ms, 200ms, 400ms...
        retry_on_exception=lambda ex: isinstance(ex, ValueError) and "版本冲突" in str(ex)
    )
    def deduct_stock(self, product_id: int, count: int) -> bool:
        """
        扣减库存（乐观锁 + 自动重试）
        """
        try:
            # 1. 查询当前商品信息（包含版本号）
            product = self.get_product(product_id)
            logger.info(f"查询商品: {product}")

            # 2. 检查库存
            if product.stock < count:
                raise ValueError(f"库存不足: 当前{product.stock}, 请求{count}")

            # 3. 计算新库存和版本号
            new_stock = product.stock - count
            old_version = product.version
            new_version = old_version + 1

            # 4. 执行乐观锁更新
            result = self.session.query(Product)\
                .filter(Product.id == product_id, Product.version == old_version)\
                .update({
                    Product.stock: new_stock,
                    Product.version: new_version
                })

            # 5. 提交事务
            self.session.commit()

            if result == 1:
                logger.info(f"✅ 扣减成功: 新库存={new_stock}, 新版本={new_version}")
                return True
            else:
                self.session.rollback()
                raise ValueError(f"版本冲突: 商品ID={product_id}, 期望版本={old_version}, 当前可能已更新")

        except Exception as e:
            self.session.rollback()
            logger.error(f"扣减库存失败: {e}")
            raise
        finally:
            self.session.close()

    def close(self):
        self.session.close()

✅ 关键点：

• @retry 装饰器自动处理重试
• filter(..., version=old_version) 实现乐观锁
• update() 返回影响行数，0 表示版本不匹配

---

5. Flask API 接口

# app.py
from flask import Flask, request, jsonify
from .services import ProductService

app = Flask(__name__)
product_service = ProductService()

@app.route('/product/<int:product_id>', methods=['GET'])
def get_product(product_id):
    try:
        product = product_service.get_product(product_id)
        return jsonify({
            'id': product.id,
            'name': product.name,
            'stock': product.stock,
            'price': float(product.price) if product.price else None,
            'version': product.version
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 400

@app.route('/product/<int:product_id>/deduct', methods=['POST'])
def deduct_stock(product_id):
    try:
        data = request.get_json()
        count = data.get('count', 1)
        
        success = product_service.deduct_stock(product_id, count)
        
        if success:
            return jsonify({'message': '库存扣减成功'})
        else:
            return jsonify({'error': '库存扣减失败'}), 400
            
    except ValueError as ve:
        return jsonify({'error': f'业务错误: {ve}'}), 400
    except Exception as e:
        return jsonify({'error': f'系统错误: {e}'}), 500

@app.route('/health', methods=['GET'])
def health():
    return jsonify({'status': 'OK'})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

---

6. 测试脚本（模拟并发）

# test_concurrent.py
import requests
import threading
import time

BASE_URL = "http://localhost:5000"

def deduct_stock(user_id, product_id, count):
    url = f"{BASE_URL}/product/{product_id}/deduct"
    payload = {"count": count}
    
    start_time = time.time()
    response = requests.post(url, json=payload)
    end_time = time.time()
    
    print(f"[用户{user_id}] 状态码: {response.status_code}, 响应: {response.json()}, 耗时: {end_time - start_time:.2f}s")

# 场景：两个用户同时扣减库存
if __name__ == "__main__":
    product_id = 1
    count = 1

    # 创建两个线程模拟并发请求
    t1 = threading.Thread(target=deduct_stock, args=(1, product_id, count))
    t2 = threading.Thread(target=deduct_stock, args=(2, product_id, count))

    t1.start()
    t2.start()

    t1.join()
    t2.join()

    # 查看最终结果
    final = requests.get(f"{BASE_URL}/product/{product_id}")
    print(f"\n最终商品状态: {final.json()}")

---

四、运行与测试

1. 启动服务

python app.py

2. 执行并发测试

python test_concurrent.py

3. 预期输出

[用户1] 状态码: 200, 响应: {'message': '库存扣减成功'}, 耗时: 0.12s
[用户2] 状态码: 400, 响应: {'error': '业务错误: 版本冲突: 商品ID=1...'}, 耗时: 0.11s

最终商品状态: {'id': 1, 'name': 'iPhone 15', 'stock': 9, 'price': 5999.0, 'version': 1}

👉 只有一个请求成功，库存正确从 10 → 9，避免了超卖。

---

五、核心优势总结

特性	说明
无阻塞读	查询性能高，适合读多写少场景
自动重试	结合 retrying 库简化开发
简单可靠	仅需一个 version 字段即可实现
易于调试	失败时明确提示“版本冲突”

---

六、生产环境建议

1. 重试策略优化：
   • 使用指数退避（Exponential Backoff）
   • 设置最大重试时间而非次数
2. 监控告警：
   • 记录乐观锁失败次数
   • 高频失败可能预示热点数据问题
3. 结合缓存：
   • 使用 Redis 预减库存，减少数据库压力
4. 分布式场景：
   • 在微服务中可通过消息队列异步处理

---

七、结语

乐观锁是构建高性能、高并发系统的基石之一。
通过 Python + SQLAlchemy 的实现，你可以轻松在 Web 应用、订单系统、库存管理等场景中应用此模式，确保数据一致性的同时最大化系统吞吐量。