Python数据API悲观锁怎么用

wen python案例 2026-07-05 84

本文目录导读：

Python数据API悲观锁怎么用

MySQL/PostgreSQL + SQLAlchemy
FastAPI + SQLAlchemy 异步支持
Redis 分布式锁（模拟悲观锁）
Django ORM中的悲观锁
最佳实践建议
完整示例：秒杀系统

在Python中实现数据API的悲观锁，主要取决于你使用的数据库和后端框架,以下是几种常见场景的实现方式：

MySQL/PostgreSQL + SQLAlchemy

基本用法

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import text
Base = declarative_base()
class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    stock = Column(Integer)
# 创建引擎和会话
engine = create_engine('mysql+pymysql://user:password@localhost/dbname')
Session = sessionmaker(bind=engine)
def update_product_stock(product_id, quantity):
    session = Session()
    try:
        # 使用FOR UPDATE悲观锁
        product = session.query(Product).filter(
            Product.id == product_id
        ).with_for_update().first()
        if product and product.stock >= quantity:
            product.stock -= quantity
            session.commit()
            return True
        return False
    except Exception as e:
        session.rollback()
        raise e
    finally:
        session.close()

更复杂的锁配置

# 带超时和跳过锁定的配置
product = session.query(Product).filter(
    Product.id == product_id
).with_for_update(
    nowait=True,  # 立即返回，不等待锁
    of=Product    # 只锁特定表
).first()
# 或者设置超时
product = session.query(Product).filter(
    Product.id == product_id
).with_for_update(
    timeout=5  # 等待5秒超时
).first()

FastAPI + SQLAlchemy 异步支持

from fastapi import FastAPI, HTTPException
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select
app = FastAPI()
# 异步引擎
async_engine = create_async_engine('mysql+aiomysql://user:password@localhost/dbname')
AsyncSessionLocal = sessionmaker(
    async_engine, 
    class_=AsyncSession, 
    expire_on_commit=False
)
@app.post("/order/{product_id}")
async def create_order(product_id: int, quantity: int):
    async with AsyncSessionLocal() as session:
        try:
            # 悲观锁查询
            stmt = select(Product).where(
                Product.id == product_id
            ).with_for_update()
            result = await session.execute(stmt)
            product = result.scalar_one_or_none()
            if not product:
                raise HTTPException(status_code=404, detail="Product not found")
            if product.stock < quantity:
                raise HTTPException(status_code=400, detail="Insufficient stock")
            product.stock -= quantity
            await session.commit()
            return {"message": "Order created successfully", "remaining": product.stock}
        except Exception as e:
            await session.rollback()
            raise

Redis 分布式锁（模拟悲观锁）

当使用Redis作为缓存时,可以实现分布式悲观锁：

import redis
from contextlib import contextmanager
redis_client = redis.Redis(host='localhost', port=6379, db=0)
class RedisDistributedLock:
    def __init__(self, lock_key, timeout=10):
        self.lock_key = f"lock:{lock_key}"
        self.timeout = timeout
    @contextmanager
    def acquire(self):
        lock_id = str(uuid.uuid4())
        try:
            # 获取锁
            acquired = redis_client.set(
                self.lock_key, 
                lock_id, 
                nx=True,           # 只在key不存在时设置
                ex=self.timeout    # 过期时间
            )
            if not acquired:
                raise Exception("Could not acquire lock")
            yield
        finally:
            # 释放锁
            self.release(lock_id)
    def release(self, lock_id):
        # 使用Lua脚本确保原子性
        lua_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        redis_client.eval(lua_script, 1, self.lock_key, lock_id)
# 使用示例
def update_stock_with_redis(product_id, quantity):
    lock = RedisDistributedLock(f"product_stock_{product_id}")
    with lock.acquire():
        # 检查库存
        current_stock = redis_client.get(f"stock:{product_id}")
        if current_stock and int(current_stock) >= quantity:
            new_stock = int(current_stock) - quantity
            redis_client.set(f"stock:{product_id}", new_stock)
            return True
        return False

Django ORM中的悲观锁

from django.db import transaction
from django.db.models import F
@transaction.atomic
def update_product_stock(product_id, quantity):
    from myapp.models import Product
    # 使用select_for_update实现悲观锁
    product = Product.objects.select_for_update().get(id=product_id)
    if product.stock >= quantity:
        # 安全的更新
        product.stock = F('stock') - quantity
        product.save(update_fields=['stock'])
        return True
    return False

最佳实践建议

合理设置事务边界

# 只包含必要的操作在事务内
@contextmanager
def pessimistic_lock_context(session, model, model_id):
    """悲观锁上下文管理器"""
    try:
        obj = session.query(model).filter(
            model.id == model_id
        ).with_for_update().first()
        yield obj
        session.commit()
    except Exception:
        session.rollback()
        raise
# 使用
with pessimistic_lock_context(session, Product, product_id) as product:
    if product:
        # 执行需要保护的操作
        pass

设置适当的超时

# 避免死锁
session.execute(
    "SELECT * FROM products WHERE id = :id FOR UPDATE NOWAIT", 
    {"id": product_id}
)

考虑性能影响

悲观锁会降低并发性能
仅在需要严格数据一致性时使用
考虑使用乐观锁或版本号替代

分布式系统考虑

使用分布式锁（Redis/ZooKeeper）
考虑使用消息队列进行异步处理

完整示例：秒杀系统

from fastapi import FastAPI, HTTPException
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.orm import sessionmaker, declarative_base
from datetime import datetime
import asyncio
app = FastAPI()
Base = declarative_base()
class SeckillItem(Base):
    __tablename__ = 'seckill_items'
    id = Column(Integer, primary_key=True)
    product_id = Column(Integer, nullable=False)
    stock = Column(Integer, default=0)
    version = Column(Integer, default=0)  # 乐观锁版本号
    start_time = Column(DateTime)
    end_time = Column(DateTime)
# 悲观锁实现
@app.post("/seckill/{item_id}")
async def seckill_with_pessimistic_lock(item_id: int, user_id: int):
    session = Session()
    try:
        # 使用悲观锁
        item = session.query(SeckillItem).filter(
            SeckillItem.id == item_id
        ).with_for_update().first()
        if not item:
            raise HTTPException(status_code=404, detail="Item not found")
        now = datetime.utcnow()
        if now < item.start_time or now > item.end_time:
            raise HTTPException(status_code=400, detail="Not in seckill time")
        if item.stock <= 0:
            raise HTTPException(status_code=400, detail="Sold out")
        # 更新库存
        item.stock -= 1
        session.commit()
        # 记录订单
        order = Order(user_id=user_id, item_id=item_id)
        session.add(order)
        session.commit()
        return {"success": True, "remaining_stock": item.stock}
    except Exception as e:
        session.rollback()
        raise HTTPException(status_code=500, detail=str(e))
    finally:
        session.close()

记住：悲观锁虽然能保证数据一致性，但会降低并发性能，在实际生产环境中，需要根据业务场景权衡使用，考虑使用乐观锁、版本号或事务隔离级别等替代方案。

本文地址： https://www.bmjup.com/post/16342.html