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