引言：Redis在现代电商架构中的核心地位
在当今高速发展的电子商务领域，系统的性能、响应速度和用户体验直接决定了平台的竞争力。Redis作为一个开源的内存数据结构存储系统，凭借其卓越的性能、丰富的数据结构和灵活的扩展能力，已经成为电商架构中不可或缺的基础设施。
Redis的核心优势在于：
极高的读写性能
：单机QPS可达10万+，通过集群扩展可支撑百万级并发
丰富的数据结构
：支持字符串、哈希、列表、集合、有序集合等多种数据结构，满足不同业务场景需求
原子操作保证
：通过Lua脚本和事务机制，确保复杂业务逻辑的数据一致性
高可用架构
：主从复制+哨兵模式或Redis Cluster，保证服务的高可用性
灵活的过期策略
：支持自动清理临时数据，节省内存空间
在电商场景中，Redis主要解决以下核心问题：
高并发场景下的性能瓶颈
：如秒杀、大促活动
复杂业务逻辑的数据一致性
：如库存扣减、订单状态管理
海量数据的实时计算
：如用户行为分析、个性化推荐
分布式系统的协调
：如分布式锁、会话管理
本文将从秒杀系统、购物车、个性化推荐、库存管理、分布式锁等核心场景，深入解析Redis在电商中的实战应用，并提供完整的代码示例和架构设计。
秒杀系统：高并发下的库存管理与防超卖
秒杀场景的挑战
秒杀是电商场景中最具挑战性的业务，具有以下特点：
瞬时流量巨大
：QPS可达平时的100-1000倍
库存有限
：通常只有几百到几千件商品
竞争激烈
：大量用户同时抢购同一商品
防超卖要求严格
：绝对不能出现库存为负的情况
Redis解决方案架构
1. 预热阶段：库存预热
在秒杀开始前，将库存数据加载到Redis中，避免直接打爆数据库：
import redis
import json
class SeckillService:
def __init__(self, redis_client):
self.redis = redis_client
self.seckill_key_prefix = “seckill:stock:”
self.seckill_order_prefix = “seckill:order:”
def preload_stock(self, item_id, stock_count):
“””
预热库存数据到Redis
item_id: 商品ID
stock_count: 库存数量
“””
key = f”{self.seckill_key_prefix}{item_id}”
# 使用Lua脚本保证原子性
lua_script = “””
local key = KEYS[1]
local stock = tonumber(ARGV[1])
local exists = redis.call(‘EXISTS’, key)
if exists == 1 then
return 0 — 库存已存在
else
redis.call(‘SET’, key, stock)
return 1 — 预热成功
end
“””
result = self.redis.eval(lua_script, 1, key, stock_count)
return result == 1
2. 秒杀核心逻辑：库存扣减
使用Redis的原子操作和Lua脚本保证库存扣减的准确性：
def seckill_purchase(self, user_id, item_id):
“””
秒杀购买逻辑
user_id: 用户ID
item_id: 商品ID
返回: (成功/失败, 订单号/错误信息)
“””
stock_key = f”{self.seckill_key_prefix}{item_id}”
order_key = f”{self.seckill_order_prefix}{item_id}:{user_id}”
# Lua脚本：原子性检查库存并扣减
lua_script = “””
local stock_key = KEYS[1]
local order_key = KEYS[2]
local user_id = ARGV[1]
local item_id = ARGV[2]
— 检查是否已购买
local exists = redis.call(‘EXISTS’, order_key)
if exists == 1 then
return {-1, “ALREADY_PURCHASED”} — 已购买
end
— 检查库存
local stock = tonumber(redis.call(‘GET’, stock_key))
if stock <= 0 then
return {-2, "OUT_OF_STOCK"} -- 库存不足
end
-- 扣减库存
redis.call('DECR', stock_key)
-- 记录订单
local order_id = user_id .. ":" .. item_id .. ":" .. redis.call('TIME')[1]
redis.call('SET', order_key, order_id)
redis.call('EXPIRE', order_key, 3600) -- 1小时过期
return {1, order_id} -- 成功
"""
try:
result = self.redis.eval(
lua_script,
2,
stock_key,
order_key,
user_id,
item_id
)
if result[0] == 1:
# 异步写入数据库（通过消息队列）
self.send_to_queue(user_id, item_id, result[1])
return True, result[1]
else:
return False, result[1]
except Exception as e:
return False, str(e)
def send_to_queue(self, user_id, item_id, order_id):
"""
将订单信息发送到消息队列，异步处理
"""
order_info = {
"user_id": user_id,
"item_id": item_id,
"order_id": order_id,
"timestamp": time.time()
}
# 这里可以接入RabbitMQ、Kafka等消息队列
# redis.lpush("seckill:order:queue", json.dumps(order_info))
print(f"订单 {order_id} 已进入处理队列")
3. 分布式限流：保护系统
使用Redis的令牌桶算法实现分布式限流：
import time
import redis
class RateLimiter:
def __init__(self, redis_client):
self.redis = redis_client
def is_allowed(self, user_id, item_id, capacity=100, refill_rate=10):
"""
令牌桶限流
capacity: 桶容量
refill_rate: 每秒填充速率
"""
key = f"rate_limit:{user_id}:{item_id}"
lua_script = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local tokens = tonumber(redis.call('GET', key) or capacity)
local last_refill = tonumber(redis.call('GET', key .. ':last') or now)
-- 计算新令牌
local elapsed = now - last_refill
local new_tokens = math.min(capacity, tokens + elapsed * refill_rate)
-- 消耗令牌
if new_tokens >= 1 then
redis.call(‘SET’, key, new_tokens – 1)
redis.call(‘SET’, key .. ‘:last’, now)
redis.call(‘EXPIRE’, key, 3600)
redis.call(‘EXPIRE’, key .. ‘:last’, 3600)
return 1
else
return 0
end
“””
now = time.time()
allowed = self.redis.eval(lua_script, 1, key, capacity, refill_rate, now)
return bool(allowed)
4. 秒杀结果查询
def get_seckill_result(self, user_id, item_id):
“””
查询秒杀结果
“””
order_key = f”{self.seckill_order_prefix}{item_id}:{user_id}”
stock_key = f”{self.seckill_key_prefix}{item_id}”
# 检查是否已购买
order_exists = self.redis.exists(order_key)
if order_exists:
order_id = self.redis.get(order_key)
return {“status”: “success”, “order_id”: order_id}
# 检查库存
stock = int(self.redis.get(stock_key) or 0)
if stock <= 0:
return {"status": "failed", "reason": "out_of_stock"}
return {"status": "pending"}
秒杀系统架构图
用户请求 → Nginx负载均衡 → 限流器(Redis) → 秒杀服务(Redis集群) → 消息队列 → 订单服务 → 数据库
关键优化点
库存预热
：提前将库存加载到Redis，避免数据库压力
原子操作
：使用Lua脚本保证库存扣减的原子性
结果异步
：订单创建异步处理，快速响应用户
限流保护
：在入口层进行流量控制
库存分段
：将库存分散到多个Redis实例，减少竞争
购物车：用户会话与数据持久化
购物车场景特点
高频读写
：用户频繁添加、删除、修改商品
数据临时性
：未登录用户数据需要临时存储
跨设备同步
：登录后需要合并多端数据
数据量大
：单个用户可能有大量商品
Redis实现方案
1. 购物车数据结构设计
使用Redis Hash存储购物车信息：
Key
:
cart:{user_id}
Field
:
item_id
Value
: JSON字符串（包含数量、属性等）
class ShoppingCartService:
def __init__(self, redis_client):
self.redis = redis_client
self.cart_key_prefix = "cart:"
self.temp_cart_prefix = "temp_cart:"
def add_to_cart(self, user_id, item_id, quantity=1, attributes=None):
"""
添加商品到购物车
user_id: 用户ID（未登录时为临时ID）
item_id: 商品ID
quantity: 数量
attributes: 商品属性（颜色、尺寸等）
"""
key = f"{self.cart_key_prefix}{user_id}"
# 商品信息
item_data = {
"item_id": item_id,
"quantity": quantity,
"attributes": attributes or {},
"added_time": time.time()
}
# 使用HSET存储
self.redis.hset(key, item_id, json.dumps(item_data))
# 设置过期时间（30天）
self.redis.expire(key, 30 * 24 * 3600)
return True
def remove_from_cart(self, user_id, item_id):
"""
从购物车移除商品
"""
key = f"{self.cart_key_prefix}{user_id}"
self.redis.hdel(key, item_id)
return True
def update_quantity(self, user_id, item_id, quantity):
"""
更新商品数量
"""
key = f"{self.cart_key_prefix}{user_id}"
if quantity <= 0:
return self.remove_from_cart(user_id, item_id)
item_json = self.redis.hget(key, item_id)
if item_json:
item_data = json.loads(item_json)
item_data["quantity"] = quantity
self.redis.hset(key, item_id, json.dumps(item_data))
return True
return False
def get_cart(self, user_id):
"""
获取购物车所有商品
"""
key = f"{self.cart_key_prefix}{user_id}"
cart_data = self.redis.hgetall(key)
result = []
total_price = 0
for item_id, item_json in cart_data.items():
item_data = json.loads(item_json)
# 这里可以查询商品实时价格
price = self.get_item_price(item_id)
item_data["price"] = price
item_data["total"] = price * item_data["quantity"]
total_price += item_data["total"]
result.append(item_data)
return {
"items": result,
"total_price": total_price,
"count": len(result)
}
def merge_carts(self, temp_id, user_id):
"""
合并临时购物车到登录用户
"""
temp_key = f"{self.temp_cart_prefix}{temp_id}"
user_key = f"{self.cart_key_prefix}{user_id}"
# 获取临时购物车数据
temp_items = self.redis.hgetall(temp_key)
for item_id, item_json in temp_items.items():
temp_data = json.loads(item_json)
# 检查用户购物车是否已有该商品
user_item_json = self.redis.hget(user_key, item_id)
if user_item_json:
user_data = json.loads(user_item_json)
# 数量相加
user_data["quantity"] += temp_data["quantity"]
self.redis.hset(user_key, item_id, json.dumps(user_data))
else:
# 直接添加
self.redis.hset(user_key, item_id, item_json)
# 删除临时购物车
self.redis.delete(temp_key)
return True
def clear_cart(self, user_id):
"""
清空购物车
"""
key = f"{self.cart_key_prefix}{user_id}"
self.redis.delete(key)
return True
2. 未登录用户购物车
使用临时ID存储，登录后合并：
def get_temp_cart_id(self, request):
"""
获取或创建临时购物车ID
"""
# 从Cookie获取
temp_id = request.cookies.get('temp_cart_id')
if not temp_id:
import uuid
temp_id = str(uuid.uuid4())
# 设置Cookie，有效期30天
response.set_cookie('temp_cart_id', temp1_id, max_age=30*24*3600)
return temp_id
3. 购物车实时价格更新
def refresh_cart_prices(self, user_id):
"""
刷新购物车商品价格
"""
key = f"{self.cart_key_prefix}{user_id}"
cart_data = self.redis.hgetall(key)
for item_id, item_json in cart_data.items():
item_data = json.loads(item_json)
# 从商品服务获取最新价格
new_price = self.get_item_price(item_id)
item_data["price"] = new_price
item_data["updated_time"] = time.time()
self.redis.hset(key, item_id, json.dumps(item_data))
购物车优化策略
数据分层
：热数据（活跃用户）保持在内存，冷数据定期清理
批量操作
：使用pipeline减少网络往返
压缩存储
：对非关键数据使用压缩算法
缓存预热
：用户登录时预加载购物车数据
个性化推荐：实时用户行为分析
推荐系统挑战
实时性
：用户行为需要立即反映在推荐结果中
海量数据
：用户行为数据量巨大
计算复杂
：协同过滤、内容推荐等算法计算量大
冷启动
：新用户和新商品缺乏数据
Redis在推荐系统中的应用
1. 用户行为实时采集
class UserBehaviorService:
def __init__(self, redis_client):
self.redis = redis_client
def record_view(self, user_id, item_id, category_id=None):
"""
记录商品浏览行为
"""
timestamp = time.time()
# 用户浏览记录（有序集合，按时间排序）
view_key = f"user:{user_id}:views"
self.redis.zadd(view_key, {item_id: timestamp})
self.redis.expire(view_key, 7 * 24 * 3600) # 保留7天
# 商品被浏览记录
item_view_key = f"item:{item_id}:viewers"
self.redis.zadd(item_view_key, {user_id: timestamp})
self.redis.expire(item_view_key, 7 * 24 * 3600)
# 分类浏览统计
if category_id:
cat_key = f"user:{user_id}:category_views"
self.redis.zincrby(cat_key, 1, category_id)
# 实时更新用户偏好向量
self.update_user_preference(user_id, item_id, "view")
def record_purchase(self, user_id, item_id, price, category_id=None):
"""
记录购买行为
"""
timestamp = time.time()
# 购买记录
purchase_key = f"user:{user_id}:purchases"
self.redis.zadd(purchase_key, {item_id: timestamp})
self.redis.expire(purchase_key, 30 * 24 * 3600)
# 商品购买记录
item_purchase_key = f"item:{item_id}:buyers"
self.redis.zadd(item_purchase_key, {user_id: timestamp})
self.redis.expire(item_purchase_key, 30 * 24 * 3600)
# 购买金额统计（用于RFM分析）
rfm_key = f"user:{user_id}:rfm"
self.redis.hincrby(rfm_key, "monetary", int(price * 100))
self.redis.hincrby(rfm_key, "frequency", 1)
self.redis.hset(rfm_key, "last_purchase", timestamp)
# 更新用户偏好（购买权重更高）
self.update_user_preference(user_id, item_id, "purchase")
# 分类购买统计
if category_id:
cat_key = f"user:{user_id}:category_purchases"
self.redis.zincrby(cat_key, 2, category_id) # 购买权重2倍
def record_cart_add(self, user_id, item_id):
"""
记录加购行为
"""
cart_key = f"user:{user_id}:cart_actions"
self.redis.zincrby(cart_key, 1, item_id)
self.redis.expire(cart_key, 7 * 24 * 3600)
# 加购权重低于购买
self.update_user_preference(user_id, item_id, "cart")
def update_user_preference(self, user_id, item_id, action_type):
"""
更新用户偏好向量（基于物品的协同过滤）
"""
# 获取商品特征（从商品服务）
item_features = self.get_item_features(item_id)
if not item_features:
return
pref_key = f"user:{user_id}:preference"
# 不同行为的权重
weights = {
"view": 1,
"cart": 3,
"purchase": 5
}
weight = weights.get(action_type, 1)
# 更新用户偏好向量
for feature, score in item_features.items():
self.redis.zincrby(pref_key, weight * score, feature)
# 限制偏好向量长度
self.redis.zremrangebyrank(pref_key, 0, -101) # 只保留前100个
2. 实时推荐算法
基于物品的协同过滤（Item-CF）
class ItemCFRecommender:
def __init__(self, redis_client):
self.redis = redis_client
def calculate_item_similarity(self, item_id1, item_id2):
"""
计算两个商品的相似度（基于共同购买用户）
"""
# 获取商品1的购买用户
buyers1_key = f"item:{item_id1}:buyers"
buyers1 = self.redis.zrange(buyers1_key, 0, -1)
# 获取商品2的购买用户
buyers2_key = f"item:{item_id2}:buyers"
buyers2 = self.redis.zrange(buyers2_key, 0, -1)
# 计算交集
common_buyers = set(buyers1) & set(buyers2)
if not common_buyers:
return 0
# 余弦相似度
similarity = len(common_buyers) / (
(len(buyers1) * len(buyers2)) ** 0.5
)
return similarity
def get_similar_items(self, item_id, top_n=10):
"""
获取相似商品（预计算）
"""
# 这里应该使用离线计算的结果
# 实时计算只用于演示
similar_items = []
# 从热门商品中选取候选
hot_items = self.redis.zrevrange("hot:items", 0, 20)
for candidate in hot_items:
if candidate == item_id:
continue
similarity = self.calculate_item_similarity(item_id, candidate)
if similarity > 0:
similar_items.append((candidate, similarity))
# 排序取TopN
similar_items.sort(key=lambda x: x[1], reverse=True)
return similar_items[:top_n]
def recommend_by_item(self, user_id, item_id, top_n=5):
“””
基于用户浏览/购买的商品进行推荐
“””
similar_items = self.get_similar_items(item_id, top_n * 2)
# 过滤用户已购买/已浏览的商品
user_purchases = self.redis.zrange(f”user:{user_id}:purchases”, 0, -1)
user_views = self.redis.zrange(f”user:{user_id}:views”, 0, -1)
excluded = set(user_purchases) | set(user_views)
recommendations = []
for item, score in similar_items:
if item not in excluded:
recommendations.append({
“item_id”: item,
“score”: score,
“reason”: f”与商品 {item_id} 相似”
})
return recommendations[:top_n]
基于用户的协同过滤（User-CF）
class UserCFRecommender:
def __init__(self, redis_client):
self.redis = redis_client
def calculate_user_similarity(self, user_id1, user_id2):
“””
计算两个用户的相似度
“””
# 获取用户的购买历史
purchases1 = set(self.redis.zrange(f”user:{user_id1}:purchases”, 0, -1))
purchases2 = set(self.redis.zrange(f”user:{user_id2}:p1urchases”, 0, -1))
if not purchases1 or not purchases2:
return 0
# 余弦相似度
common_items = purchases1 & purchases2
similarity = len(common_items) / (
(len(purchases1) * len(purchases2)) ** 0.5
)
return similarity
def find_similar_users(self, user_id, top_n=20):
“””
寻找相似用户（基于购买历史）
“””
# 获取所有活跃用户
active_users = self.redis.zrange(“active:users”, 0, 1000)
similarities = []
for other_user in active_users:
if other_user == user_id:
continue
sim = self.calculate_user_similarity(user_id, other_user)
if sim > 0:
similarities.append((other_user, sim))
similarities.sort(key=lambda x: x[1], reverse=True)
return similarities[:top_n]
def recommend_by_user(self, user_id, top_n=5):
“””
基于相似用户的购买历史推荐
“””
similar_users = self.find_similar_users(user_id, 20)
# 获取相似用户的购买记录
candidate_items = {}
for similar_user, similarity in similar_users:
purchases = self.redis.zrange(f”user:{similar_user}:purchases”, 0, -1)
for item in purchases:
candidate_items[item] = candidate_items.get(item, 0) + similarity
# 过滤用户已购买的商品
user_purchases = set(self.redis.zrange(f”user:{user_id}:purchases”, 0, -1))
recommendations = []
for item_id, score in candidate_items.items():
if item_id not in user_purchases:
recommendations.append({
“item_id”: item_id,
“score”: score,
“reason”: “相似用户购买”
})
recommendations.sort(key=lambda x: x[“score”], reverse=True)
return recommendations[:top_n]
3. 实时热门推荐
class HotRecommender:
def __init__(self, redis_client):
self.redis = redis_client
def update_hot_items(self, item_id, score=1):
“””
实时更新热门商品（滑动窗口）
“””
key = “hot:items:minute” # 分钟级热度
timestamp = int(time.time() / 60) # 分钟时间戳
# 使用有序集合，score为时间戳，member为商品ID
self.redis.zincrby(key, score, item_id)
# 设置过期时间（保留最近60分钟）
self.redis.expire(key, 3600)
# 更新小时级热度
hour_key = “hot:items:hour”
self.redis.zincrby(hour_key, score, item_id)
self.redis.expire(hour_key, 24 * 3600)
def get_hot_items(self, time_window=”hour”, top_n=10):
“””
获取热门商品
“””
key = f”hot:items:{time_window}”
return self.redis.zrevrange(key, 0, top_n – 1, withscores=True)
def get_recommendations(self, user_id, top_n=5):
“””
基于热门商品推荐（用于新用户）
“””
# 获取热门商品
hot_items = self.get_hot_items(“hour”, top_n * 2)
# 过滤用户已浏览的商品
user_views = set(self.redis.zrange(f”user:{user_id}:views”, 0, -1))
recommendations = []
for item_id, score in hot_items:
if item_id not in user_views:
recommendations.append({
“item_id”: item_id,
“score”: score,
“reason”: “热门商品”
})
return recommendations[:top_n]
4. 冷启动推荐
def cold_start_recommend(self, user_id, top_n=5):
“””
新用户冷启动推荐
“””
# 1. 热门商品
hot_items = self.redis.zrevrange(“hot:items:hour”, 0, top_n – 1)
# 2. 新品推荐
new_items = self.redis.zrange(“new:items”, 0, top_n – 1)
# 3. 促销商品
promo_items = self.redis.zrange(“promo:items”, 0, top_n – 1)
# 合并并去重
recommendations = list(set(hot_items + new_items + promo_items))
return recommendations[:top_n]
推荐系统优化
预计算
：相似度矩阵离线计算，Redis存储结果
多级缓存
：热门推荐缓存、用户偏好缓存、结果缓存
A/B测试
：使用Redis存储实验配置和结果
实时更新
：用户行为实时更新偏好向量
库存管理：分布式库存扣减
库存管理挑战
分布式环境
：多实例部署，数据一致性难保证
超卖问题
：高并发下容易出现库存为负
回滚复杂
：订单取消时库存回滚
性能要求
：库存查询和扣减需要高性能
Redis解决方案
1. 分布式库存扣减
class InventoryService:
def __init__(self, redis_client):
self.redis = redis_client
def deduct_stock(self, item_id, quantity, order_id):
“””
分布式库存扣减
“””
stock_key = f”inventory:{item_id}:stock”
lock_key = f”inventory:{item_id}:lock”
# Lua脚本保证原子性
lua_script = “””
local stock_key = KEYS[1]
local lock_key = KEYS[2]
local quantity = tonumber(ARGV[1])
local order_id = ARGV[2]
local expire_time = tonumber(ARGV[3])
— 检查库存
local stock = tonumber(redis.call(‘GET’, stock_key))
if stock < quantity then
return {-1, "INSUFFICIENT_STOCK"}
end
-- 检查是否已有锁（防止重复扣减）
local lock = redis.call('GET', lock_key)
if lock and lock ~= order_id then
return {-2, "LOCKED"}
end
-- 扣减库存
redis.call('DECRBY', stock_key, quantity)
-- 设置扣减记录锁（防止重复扣减）
redis.call('SET', lock_key, order_id)
redis.call('EXPIRE', lock_key, expire_time)
-- 记录扣减日志
local log_key = "inventory:deduct:log:" .. order_id
redis.call('HSET', log_key, 'item_id', item_id)
redis.call('HSET', log_key, 'quantity', quantity)
redis.call('HSET', log_key, 'timestamp', redis.call('TIME')[1])
redis.call('EXPIRE', log_key, 24 * 3600)
return {1, stock - quantity}
"""
try:
result = self.redis.eval(
lua_script,
2,
stock_key,
lock_key,
quantity,
order_id,
3600 # 锁过期时间1小时
)
if result[0] == 1:
# 扣减成功，发送消息到MQ，异步更新数据库
self.send_stock_deduct_message(item_id, quantity, order_id)
return True, result[1]
else:
return False, result[1]
except Exception as e:
return False, str(e)
def restore_stock(self, item_id, quantity, order_id):
"""
库存回滚（订单取消时调用）
"""
stock_key = f"inventory:{item_id}:stock"
lock_key = f"inventory:{item_id}:lock"
# 检查锁是否匹配
current_lock = self.redis.get(lock_key)
if current_lock != order_id:
return False, "LOCK_MISMATCH"
# 回滚库存
self.redis.incrby(stock_key, quantity)
# 删除锁
self.redis.delete(lock_key)
# 记录回滚日志
log_key = f"inventory:restore:log:{order_id}"
self.redis.hset(log_key, "item_id", item_id)
self.redis.hset(log_key, "quantity", quantity)
self.redis.hset(log_key, "timestamp", time.time())
# 发送回滚消息
self.send_stock_restore_message(item_id, quantity, order_id)
return True, "RESTORED"
def get_stock(self, item_id):
"""
获取实时库存
"""
stock_key = f"inventory:{item_id}:stock"
stock = self.redis.get(stock_key)
if stock is None:
# 缓存未命中，从数据库加载
stock = self.load_stock_from_db(item_id)
if stock is not None:
self.redis.set(stock_key, stock)
self.redis.expire(stock_key, 60) # 1分钟过期
return int(stock) if stock else 0
def batch_deduct_stock(self, items, order_id):
"""
批量扣减库存（购物车结算）
"""
# 使用pipeline批量操作
pipe = self.redis.pipeline()
for item_id, quantity in items.items():
stock_key = f"inventory:{item_id}:stock"
pipe.get(stock_key)
stocks = pipe.execute()
# 检查所有商品库存
for i, (item_id, quantity) in enumerate(items.items()):
if int(stocks[i]) < quantity:
return False, f"商品 {item_id} 库存不足"
# 批量扣减
deduct_pipe = self.redis.pipeline()
for item_id, quantity in items.items():
stock_key = f"inventory:{item_id}:stock"
deduct_pipe.decrby(stock_key, quantity)
deduct_pipe.execute()
# 记录批量扣减
log_key = f"inventory:batch:log:{order_id}"
deduct_pipe.hset(log_key, "items", json.dumps(items))
deduct_pipe.hset(log_key, "timestamp", time.time())
deduct_pipe.expire(log_key, 24 * 3600)
deduct_pipe.execute()
return True, "SUCCESS"
2. 库存预警系统
class InventoryAlertService:
def __init__(self, redis_client):
self.redis = redis_client
def check_low_stock(self, item_id, threshold=100):
"""
检查库存是否低于阈值
"""
stock_key = f"inventory:{item_id}:stock"
stock = int(self.redis.get(stock_key) or 0)
if stock < threshold:
# 发送预警消息
alert_key = "inventory:alert:low"
alert_data = {
"item_id": item_id,
"stock": stock,
"threshold": threshold,
"timestamp": time.time()
}
self.redis.lpush(alert_key, json.dumps(alert_data))
# 触发预警处理（可以接入消息队列）
self.trigger_alert_process(alert_data)
return True
return False
def batch_check_low_stock(self, threshold=100):
"""
批量检查低库存商品
"""
# 获取所有库存商品
pattern = "inventory:*:stock"
stock_keys = self.redis.keys(pattern)
low_stock_items = []
for key in stock_keys:
item_id = key.decode().split(":")[1]
if self.check_low_stock(item_id, threshold):
low_stock_items.append(item_id)
return low_stock_items
def get_stock_status(self, item_id):
"""
获取库存状态
"""
stock_key = f"inventory:{item_id}:stock"
stock = int(self.redis.get(stock_key) or 0)
if stock == 0:
return "OUT_OF_STOCK"
elif stock < 10:
return "LOW_STOCK"
elif stock < 100:
return "MEDIUM_STOCK"
else:
return "HIGH_STOCK"
3. 库存预热与同步
def preload_inventory(self, item_ids):
"""
批量预热库存到Redis
"""
# 从数据库加载库存
inventory_data = self.load_inventory_from_db(item_ids)
# 批量写入Redis
pipe = self.redis.pipeline()
for item_id, stock in inventory_data.items():
key = f"inventory:{item_id}:stock"
pipe.set(key, stock)
pipe.expire(key, 3600) # 1小时过期
pipe.execute()
return len(inventory_data)
def sync_inventory_to_db(self, item_id):
"""
同步Redis库存到数据库（定时任务）
"""
stock_key = f"inventory:{item_id}:stock"
current_stock = int(self.redis.get(stock_key) or 0)
# 更新数据库
self.update_db_stock(item_id, current_stock)
# 记录同步日志
log_key = "inventory:sync:log"
self.redis.lpush(log_key, json.dumps({
"item_id": item_id,
"stock": current_stock,
"timestamp": time.time()
}))
# 限制日志长度
self.redis.ltrim(log_key, 0, 999)
库存管理优化
分段库存
：将库存分散到多个key，减少竞争
缓存预热
：活动前加载库存到Redis
异步同步
：库存变更异步同步到数据库
监控告警
：实时监控库存水位
分布式锁：保证数据一致性
分布式锁场景
库存扣减
：防止超卖
订单创建
：防止重复订单
优惠券领取
：防止超领
支付回调
：防止重复处理
Redis分布式锁实现
1. 基础分布式锁
import time
import uuid
class DistributedLock:
def __init__(self, redis_client):
self.redis = redis_client
self.lock_timeout = 30 # 锁超时时间（秒）
def acquire_lock(self, lock_key, timeout=None):
"""
获取分布式锁
lock_key: 锁键
timeout: 锁超时时间（秒）
返回: lock_value（成功时）或None（失败）
"""
if timeout is None:
timeout = self.lock_timeout
lock_value = str(uuid.uuid4())
end_time = time.time() + timeout
while time.time() < end_time:
# SET NX EX 原子操作
acquired = self.redis.set(
lock_key,
lock_value,
nx=True, # 只有key不存在时才设置
ex=timeout # 自动过期
)
if acquired:
return lock_value
# 短暂等待后重试
time.sleep(0.001)
return None
def release_lock(self, lock_key, lock_value):
"""
释放分布式锁（使用Lua脚本保证原子性）
"""
lua_script = """
local key = KEYS[1]
local value = ARGV[1]
local current_value = redis.call('GET', key)
if current_value == value then
redis.call('DEL', key)
return 1
else
return 0
end
"""
result = self.redis.eval(lua_script, 1, lock_key, lock_value)
return bool(result)
def extend_lock(self, lock_key, lock_value, additional_time):
"""
续期锁（防止业务执行时间过长）
"""
lua_script = """
local key = KEYS[1]
local value = ARGV[1]
local expire = tonumber(ARGV[2])
local current_value = redis.call('GET', key)
if current_value == value then
redis.call('EXPIRE', key, expire)
return 1
else
return 0
end
"""
result = self.redis.eval(
lua_script,
1,
lock_key,
lock_value,
additional_time
)
return bool(result)
2. 可重入锁实现
class ReentrantDistributedLock:
def __init__(self, redis_client):
self.redis = redis_client
def acquire_lock(self, lock_key, lock_value, timeout=30):
"""
可重入锁：同一个线程可以多次获取同一把锁
"""
# 使用hash存储锁信息：key=lock_key, field=thread_id, value=count
lock_info_key = f"lock:info:{lock_key}"
thread_id = str(uuid.uuid4()) # 实际应用中应使用线程ID
lua_script = """
local lock_info_key = KEYS[1]
local thread_id = ARGV[1]
local timeout = tonumber(ARGV[2])
local current_time = tonumber(ARGV[3])
-- 检查是否已持有锁
local count = tonumber(redis.call('HGET', lock_info_key, thread_id) or 0)
if count > 0 then
— 可重入：计数器+1
redis.call(‘HINCRBY’, lock_info_key, thread_id, 1)
redis.call(‘EXPIRE’, lock_info_key, timeout)
return {1, thread_id, count + 1}
end
— 尝试获取锁
local lock_key = KEYS[2]
local lock_value = ARGV[4]
local acquired = redis.call(‘SET’, lock_key, lock_value, ‘NX’, ‘EX’, timeout)
if acquired then
redis.call(‘HSET’, lock_info_key, thread_id, 1)
redis.call(‘EXPIRE’, lock_info_key, timeout)
return {1, thread_id, 1}
end
return {0, nil, 0}
“””
lock_value = str(uuid.uuid4())
result = self.redis.eval(
lua_script,
2,
lock_info_key,
lock_key,
thread_id,
timeout,
time.time(),
lock_value
)
if result[0] == 1:
return {
“lock_value”: lock_value,
“thread_id”: result[1],
“count”: result[2]
}
return None
def release_lock(self, lock_key, lock_info):
“””
释放可重入锁
“””
lock_info_key = f”lock:info:{lock_key}”
thread_id = lock_info[“thread_id”]
lock_value = lock_info[“lock_value”]
lua_script = “””
local lock_info_key = KEYS[1]
local lock_key = KEYS[2]
local thread_id = ARGV[1]
local lock_value = ARGV[2]
— 检查锁是否匹配
local current_value = redis.call(‘GET’, lock_key)
if current_value ~= lock_value then
return 0
end
— 减少计数器
local count = tonumber(redis.call(‘HGET’, lock_info_key, thread_id) or 0)
if count > 1 then
— 还有重入次数
redis.call(‘HINCRBY’, lock_info_key, thread_id, -1)
return 2
elseif count == 1 then
— 最后一次，删除锁
redis.call(‘HDEL’, lock_info_key, thread_id)
redis.call(‘DEL’, lock_key)
return 1
else
— 异常情况
return 0
end
“””
result = self.redis.eval(
lua_script,
2,
lock_info_key,
lock_key,
thread_id,
lock_value
)
return result
3. Redlock算法实现（多节点锁）
class Redlock:
def __init__(self, redis_nodes):
“””
redis_nodes: Redis连接列表，至少3个节点
“””
self.redis_nodes = redis_nodes
self.clock_drift_factor = 0.01
self.retry_count = 3
self.retry_delay = 200 # ms
def acquire_lock(self, lock_key, validity_time=30000):
“””
Redlock算法获取锁
validity_time: 锁的有效期（毫秒）
“””
lock_value = str(uuid.uuid4())
quorum = len(self.redis_nodes) // 2 + 1
start_time = time.time() * 1000
for retry in range(self.retry_count):
success_count = 0
# 向所有节点请求锁
for node in self.redis_nodes:
try:
acquired = node.set(
lock_key,
lock_value,
nx=True,
px=validity_time
)
if acquired:
success_count += 1
except Exception:
pass
# 检查是否获得多数派锁
end_time = time.time() * 1000
elapsed_time = end_time – start_time
# 有效时间减去时钟漂移
validity_time_remaining = validity_time – elapsed_time – 2
if success_count >= quorum and validity_time_remaining > 0:
return {
“lock_value”: lock_value,
“validity_time”: validity_time_remaining
}
# 释放已获得的锁
self.release_lock(lock_key, lock_value)
# 等待后重试
if retry < self.retry_count - 1:
time.sleep(self.retry_delay / 1000)
return None
def release_lock(self, lock_key, lock_value):
"""
释放Redlock
"""
for node in self.redis_nodes:
try:
lua_script = """
local key = KEYS[1]
local value = ARGV[1]
local current_value = redis.call('GET', key)
if current_value == value then
redis.call('DEL', key)
return 1
end
return 0
"""
node.eval(lua_script, 1, lock_key, lock_value)
except Exception:
pass
4. 分布式锁使用示例
# 库存扣减使用分布式锁
def deduct_stock_with_lock(self, item_id, quantity, order_id):
lock_key = f"lock:inventory:{item_id}"
lock = self.distributed_lock.acquire_lock(lock_key)
if not lock:
return False, "获取锁失败"
try:
# 执行库存扣减
stock_key = f"inventory:{item_id}:stock"
current_stock = int(self.redis.get(stock_key) or 0)
if current_stock < quantity:
return False, "库存不足"
self.redis.decrby(stock_key, quantity)
# 记录订单
order_key = f"order:{order_id}"
self.redis.set(order_key, json.dumps({
"item_id": item_id,
"quantity": quantity,
"status": "created"
}))
return True, "SUCCESS"
finally:
# 释放锁
self.distributed_lock.release_lock(lock_key, lock)
分布式锁最佳实践
设置合理的超时时间
：防止死锁
锁续期机制
：长任务时自动续期
避免锁嵌套
：减少死锁风险
监控锁状态
：记录锁的获取/释放日志
会话管理：分布式Session存储
Session管理挑战
分布式部署
：多台服务器需要共享Session
性能要求
：Session操作频繁，需要高性能存储
安全性
：防止Session劫持
过期管理
：自动清理过期Session
Redis实现方案
1. Session存储结构
class SessionService:
def __init__(self, redis_client):
self.redis = redis_client
self.session_timeout = 3600 # 1小时
def create_session(self, user_id, user_info):
"""
创建Session
"""
session_id = str(uuid.uuid4())
session_key = f"session:{session_id}"
session_data = {
"user_id": user_id,
"user_info": user_info,
"created_at": time.time(),
"last_activity": time.time(),
"ip": user_info.get("ip", ""),
"user_agent": user_info.get("user_agent", "")
}
# 存储Session数据
self.redis.hset(session_key, mapping={
k: json.dumps(v) if isinstance(v, (dict, list)) else str(v)
for k, v in session_data.items()
})
# 设置过期时间
self.redis.expire(session_key, self.session_timeout)
# 记录用户Session索引（用于用户踢出）
user_session_key = f"user:{user_id}:sessions"
self.redis.sadd(user_session_key, session_id)
self.redis.expire(user_session_key, self.session_timeout)
return session_id
def get_session(self, session_id):
"""
获取Session
"""
session_key = f"session:{session_id}"
session_data = self.redis.hgetall(session_key)
if not session_data:
return None
# 解析数据
session = {}
for k, v in session_data.items():
k = k.decode() if isinstance(k, bytes) else k
v = v.decode() if isinstance(v, bytes) else v
try:
session[k] = json.loads(v)
except:
session[k] = v
# 更新最后活动时间
self.redis.hset(session_key, "last_activity", time.time())
self.redis.expire(session_key, self.session_timeout)
return session
def update_session(self, session_id, updates):
"""
更新Session
"""
session_key = f"session:{session_id}"
# 检查Session是否存在
if not self.redis.exists(session_key):
return False
# 更新数据
for k, v in updates.items():
if isinstance(v, (dict, list)):
self.redis.hset(session_key, k, json.dumps(v))
else:
self.redis.hset(session_key, k, str(v))
# 更新最后活动时间
self.redis.hset(session_key, "last_activity", time.time())
self.redis.expire(session_key, self.session_timeout)
return True
def destroy_session(self, session_id):
"""
销毁Session
"""
session_key = f"session:{session_id}"
# 获取用户ID
user_id = self.redis.hget(session_key, "user_id")
if user_id:
user_id = user_id.decode() if isinstance(user_id, bytes) else user_id
user_session_key = f"user:{user_id}:sessions"
self.redis.srem(user_session_key, session_id)
# 删除Session
self.redis.delete(session_key)
return True
def destroy_all_user_sessions(self, user_id):
"""
强制用户下线（踢出用户）
"""
user_session_key = f"user:{user_id}:sessions"
session_ids = self.redis.smembers(user_session_key)
for session_id in session_ids:
session_id = session_id.decode() if isinstance(session_id, bytes) else session_id
self.destroy_session(session_id)
self.redis.delete(user_session_key)
return True
def get_user_sessions(self, user_id):
"""
获取用户所有Session
"""
user_session_key = f"user:{user_id}:sessions"
session_ids = self.redis.smembers(user_session_key)
sessions = []
for session_id in session_ids:
session_id = session_id.decode() if isinstance(session_id, bytes) else session_id
session = self.get_session(session_id)
if session:
sessions.append(session)
return sessions
2. Session清理与监控
def cleanup_expired_sessions(self):
"""
清理过期Session（定时任务）
"""
# 使用scan迭代所有Session key
pattern = "session:*"
cursor = 0
cleaned = 0
while True:
cursor, keys = self.redis.scan(cursor, match=pattern, count=100)
for key in keys:
key = key.decode() if isinstance(key, bytes) else key
# 检查是否过期（通过TTL）
ttl = self.redis.ttl(key)
if ttl == -1: # 没有设置过期时间
self.redis.delete(key)
cleaned += 1
if cursor == 0:
break
return cleaned
def get_session_stats(self):
"""
获取Session统计信息
"""
pattern = "session:*"
total_sessions = 0
active_sessions = 0
cursor = 0
while True:
cursor, keys = self.redis.scan(cursor, match=pattern, count=1000)
total_sessions += len(keys)
for key in keys:
last_activity = self.redis.hget(key, "last_activity")
if last_activity:
last_activity = float(last_activity.decode())
if time.time() - last_activity < 1800: # 30分钟内活跃
active_sessions += 1
if cursor == 0:
break
return {
"total_sessions": total_sessions,
"active_sessions": active_sessions,
"inactive_sessions": total_sessions - active_sessions
}
3. Session安全增强
class SecureSessionService(SessionService):
def __init__(self, redis_client):
super().__init__(redis_client)
self.max_sessions_per_user = 5 # 单用户最多Session数
def create_session(self, user_id, user_info):
"""
创建Session（带安全检查）
"""
# 检查用户Session数量
user_session_key = f"user:{user_id}:sessions"
current_sessions = self.redis.scard(user_session_key)
if current_sessions >= self.max_sessions_per_user:
# 删除最旧的Session
oldest_session = self.redis.srandmember(user_session_key)
if oldest_session:
self.destroy_session(oldest_session)
# 创建新Session
session_id = super().create_session(user_id, user_info)
# 记录登录日志
self.log_login(user_id, user_info)
return session_id
def log_login(self, user_id, user_info):
“””
记录登录日志
“””
log_key = f”security:login_log:{user_id}”
log_entry = {
“timestamp”: time.time(),
“ip”: user_info.get(“ip”, “”),
“user_agent”: user_info.get(“user_agent”, “”),
“action”: “login”
}
self.redis.lpush(log_key, json.dumps(log_entry))
self.redis.ltrim(log_key, 0, 99) # 保留最近100条
self.redis.expire(log_key, 30 * 24 * 3600) # 30天
def validate_session(self, session_id, request_ip, request_ua):
“””
验证Session（防劫持）
“””
session = self.get_session(session_id)
if not session:
return False
# 检查IP是否变化
if session.get(“ip”) != request_ip:
# 记录异常
self.log_security_event(“ip_change”, session_id, request_ip)
return False
# 检查User-Agent是否变化
if session.get(“user_agent”) != request_ua:
self.log_security_event(“ua_change”, session_id, request_ua)
return False
return True
def log_security_event(self, event_type, session_id, value):
“””
记录安全事件
“””
log_key = “security:events”
log_entry = {
“timestamp”: time.time(),
“event_type”: event_type,
“session_id”: session_id,
“value”: value
}
self.redis.lpush(log_key, json.dumps(log_entry))
self.redis.ltrim(log_key, 0, 999)
Session管理优化
Pipeline批量操作
：减少网络往返
压缩存储
：对大型Session数据进行压缩
读写分离
：主库写，从库读
监控告警
：Session数量异常告警
缓存策略：多级缓存架构
缓存挑战
数据一致性
：缓存与数据库一致性
缓存穿透
：查询不存在的数据
缓存击穿
：热点key过期
缓存雪崩
：大量key同时过期
Redis缓存解决方案
1. 缓存穿透防护
class CacheService:
def __init__(self, redis_client):
self.redis = redis_client
def get_with_cache(self, key, fallback_func, ttl=3600, null_ttl=60):
“””
布隆过滤器+空值缓存防穿透
“””
# 1. 先查布隆过滤器（判断key是否存在）
if not self.bloom_filter_check(key):
# 不存在直接返回
return None
# 2. 查缓存
cached = self.redis.get(key)
if cached:
if cached == “NULL”:
return None
return json.loads(cached)
# 3. 查数据库
result = fallback_func()
if result is None:
# 缓存空值（短过期时间）
self.redis.setex(key, null_ttl, “NULL”)
return None
else:
# 缓存真实数据
self.redis.setex(key, ttl, json.dumps(result))
return result
def bloom_filter_check(self, key):
“””
布隆过滤器检查（简化版，实际可用RedisBloom模块）
“””
# 这里使用简单的SET模拟布隆过滤器
# 实际应使用RedisBloom的BF.ADD/BF.EXISTS
bloom_key = “bloom:filter”
return self.redis.sismember(bloom_key, key)
def bloom_filter_add(self, key):
“””
添加到布隆过滤器
“””
bloom_key = “bloom:filter”
self.redis.sadd(bloom_key, key)
2. 缓存击穿防护
def get_hot_key_with_mutex(self, key, fallback_func, ttl=3600):
“””
互斥锁防击穿
“””
# 1. 查缓存
cached = self.redis.get(key)
if cached:
return json.loads(cached)
# 2. 获取分布式锁
lock_key = f”lock:{key}”
lock = self.distributed_lock.acquire_lock(lock_key, timeout=10)
if lock:
try:
# 双重检查（防止其他请求已更新缓存）
cached = self.redis.get(key)
if cached:
return json.loads(cached)
# 查询数据库
result = fallback_func()
if result:
self.redis.setex(key, ttl, json.dumps(result))
return result
finally:
self.distributed_lock.release_lock(lock_key, lock)
else:
# 未获取锁，短暂等待后重试
time.sleep(0.01)
return self.get_hot_key_with_mutex(key, fallback_func, ttl)
3. 缓存雪崩防护
def set_with_random_ttl(self, key, value, base_ttl=3600, variance=600):
“””
随机过期时间防雪崩
“””
ttl = base_ttl + random.randint(-variance, variance)
self.redis.setex(key, ttl, json.dumps(value))
def batch_set_with_staggered_ttl(self, items, base_ttl=3600):
“””
批量设置错开过期时间
“””
pipe = self.redis.pipeline()
for key, value in items.items():
ttl = base_ttl + random.randint(0, 600) # 0-10分钟随机偏移
pipe.setex(key, ttl, json.dumps(value))
pipe.execute()
4. 多级缓存架构
class MultiLevelCache:
def __init__(self, redis_client, local_cache=None):
self.redis = redis_client
self.local_cache = local_cache or {} # 本地缓存（如Caffeine）
def get(self, key, fallback_func, levels=[“local”, “redis”, “db”]):
“””
多级缓存查询
“””
result = None
# 1. 本地缓存
if “local” in levels and key in self.local_cache:
return self.local_cache[key]
# 2. Redis缓存
if “redis” in levels:
cached = self.redis.get(key)
if cached:
result = json.loads(cached)
# 回填本地缓存
self.local_cache[key] = result
return result
# 3. 数据库
if “db” in levels:
result = fallback_func()
if result:
# 回填Redis和本地缓存
self.redis.setex(key, 3600, json.dumps(result))
self.local_cache[key] = result
return result
def invalidate(self, key):
“””
失效多级缓存
“””
# 删除本地缓存
self.local_cache.pop(key, None)
# 删除Redis缓存
self.redis.delete(key)
缓存最佳实践
缓存预热
：活动前加载热点数据
热点识别
：使用Redis监控热点key
缓存分层
：本地缓存+Redis缓存+数据库
监控告警
：缓存命中率、延迟监控
实战案例：完整电商秒杀系统
系统架构设计
用户层 → 网关层(Nginx+Lua) → 业务层(微服务) → 数据层(Redis+MySQL)
完整代码实现
import redis
import json
import time
import uuid
import random
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
class EcommerceSeckillSystem:
def __init__(self, redis_client):
self.redis = redis_client
self.distributed_lock = DistributedLock(redis_client)
self.rate_limiter = RateLimiter(redis_client)
# Key前缀
self.prefixes = {
“stock”: “seckill:stock:”,
“order”: “seckill:order:”,
“user_orders”: “seckill:user_orders:”,
“lock”: “seckill:lock:”,
“rate_limit”: “seckill:rate_limit:”,
“result”: “seckill:result:”
}
def init_seckill_activity(self, item_id, stock_count, start_time, end_time):
“””
初始化秒杀活动
“””
# 1. 预热库存
stock_key = self.prefixes[“stock”] + str(item_id)
self.redis.set(stock_key, stock_count)
# 2. 设置活动时间窗口
activity_key = f”seckill:activity:{item_id}”
self.redis.hset(activity_key, mapping={
“start_time”: start_time,
“end_time”: end_time,
“total_stock”: stock_count,
“status”: “ready”
})
# 3. 初始化统计
stats_key = f”seckill:stats:{item_id}”
self.redis.hset(stats_key, mapping={
“total_requests”: 0,
“success_orders”: 0,
“failed_orders”: 0,
“start_time”: time.time()
})
return True
def seckill(self, user_id, item_id):
“””
秒杀主流程
“””
# 1. 活动状态检查
if not self.check_activity_status(item_id):
return {“success”: False, “message”: “活动未开始或已结束”}
# 2. 限流检查
rate_key = self.prefixes[“rate_limit”] + f”{user_id}:{item_id}”
if not self.rate_limiter.is_allowed(user_id, item_id, capacity=5, refill_rate=1):
return {“success”: False, “message”: “请求过于频繁，请稍后再试”}
# 3. 重复购买检查
order_key = self.prefixes[“order”] + f”{item_id}:{user_id}”
if self.redis.exists(order_key):
order_id = self.redis.get(order_key)
return {“success”: False, “message”: “您已购买成功，请勿重复购买”, “order_id”: order_id}
# 4. 获取分布式锁
lock_key = self.prefixes[“lock”] + str(item_id)
lock = self.distributed_lock.acquire_lock(lock_key, timeout=5)
if not lock:
return {“success”: False, “message”: “系统繁忙，请稍后再试”}
try:
# 5. 库存检查与扣减（原子操作）
stock_key = self.prefixes[“stock”] + str(item_id)
lua_script = “””
local stock_key = KEYS[1]
local order_key = KEYS[2]
local user_id = ARGV[1]
local item_id = ARGV[2]
— 检查是否已购买
if redis.call(‘EXISTS’, order_key) == 1 then
return {-1, “ALREADY_PURCHASED”}
end
— 检查库存
local stock = tonumber(redis.call(‘GET’, stock_key))
if stock <= 0 then
return {-2, "OUT_OF_STOCK"}
end
-- 扣减库存
redis.call('DECR', stock_key)
-- 生成订单
local order_id = user_id .. ":" .. item_id .. ":" .. redis.call('TIME')[1]
redis.call('SET', order_key, order_id)
redis.call('EXPIRE', order_key, 3600)
-- 记录用户订单
local user_orders_key = "seckill:user_orders:" .. user_id
redis.call('ZADD', user_orders_key, redis.call('TIME')[1], order_id)
redis.call('EXPIRE', user_orders_key, 24 * 3600)
return {1, order_id}
"""
result = self.redis.eval(
lua_script,
2,
stock_key,
order_key,
user_id,
item_id
)
if result[0] == 1:
# 6. 更新统计
self.update_stats(item_id, "success")
# 7. 发送消息到队列（异步创建订单）
self.send_to_queue(user_id, item_id, result[1])
# 8. 缓存秒杀结果
result_key = self.prefixes["result"] + f"{item_id}:{user_id}"
self.redis.setex(result_key, 3600, json.dumps({
"status": "success",
"order_id": result[1],
"timestamp": time.time()
}))
return {
"success": True,
"order_id": result[1],
"message": "秒杀成功"
}
else:
self.update_stats(item_id, "failed")
return {
"success": False,
"message": result[1]
}
finally:
# 9. 释放锁
self.distributed_lock.release_lock(lock_key, lock)
def check_activity_status(self, item_id):
"""
检查活动状态
"""
activity_key = f"seckill:activity:{item_id}"
activity = self.redis.hgetall(activity_key)
if not activity:
return False
start_time = float(activity.get(b"start_time", 0))
end_time = float(activity.get(b"end_time", 0))
status = activity.get(b"status", b"ready").decode()
now = time.time()
return start_time <= now <= end_time and status == "ready"
def update_stats(self, item_id, result_type):
"""
更新统计信息
"""
stats_key = f"seckill:stats:{item_id}"
if result_type == "success":
self.redis.hincrby(stats_key, "success_orders", 1)
else:
self.redis.hincrby(stats_key, "failed_orders", 1)
self.redis.hincrby(stats_key, "total_requests", 1)
def send_to_queue(self, user_id, item_id, order_id):
"""
发送到消息队列（异步处理）
"""
message = {
"user_id": user_id,
"item_id": item_id,
"order_id": order_id,
"timestamp": time.time()
}
# 实际使用RabbitMQ/Kafka
# self.message_queue.publish("seckill.orders", json.dumps(message))
print(f"发送到队列: {message}")
def get_seckill_result(self, user_id, item_id):
"""
查询秒杀结果
"""
result_key = self.prefixes["result"] + f"{item_id}:{user_id}"
result = self.redis.get(result_key)
if result:
return json.loads(result)
# 检查订单是否存在
order_key = self.prefixes["order"] + f"{item_id}:{user_id}"
if self.redis.exists(order_key):
order_id = self.redis.get(order_key)
return {
"status": "success",
"order_id": order_id
}
return {"status": "pending"}
def get_activity_stats(self, item_id):
"""
获取活动实时统计
"""
stats_key = f"seckill:stats:{item_id}"
stock_key = self.prefixes["stock"] + str(item_id)
stats = self.redis.hgetall(stats_key)
current_stock = int(self.redis.get(stock_key) or 0)
return {
"total_requests": int(stats.get(b"total_requests", 0)),
"success_orders": int(stats.get(b"success_orders", 0)),
"failed_orders": int(stats.get(b"failed_orders", 0)),
"current_stock": current_stock,
"success_rate": round(
int(stats.get(b"success_orders", 0)) /
max(int(stats.get(b"total_requests", 1)), 1) * 100,
2
)
}
def close_seckill_activity(self, item_id):
"""
关闭秒杀活动
"""
activity_key = f"seckill:activity:{item_id}"
self.redis.hset(activity_key, "status", "closed")
# 清理临时数据（可选）
# self.redis.delete(self.prefixes["stock"] + str(item_id))
return True
def batch_init_activities(self, activities):
"""
批量初始化活动
"""
pipe = self.redis.pipeline()
for activity in activities:
item_id = activity["item_id"]
stock = activity["stock"]
start = activity["start_time"]
end = activity["end_time"]
# 库存
pipe.set(self.prefixes["stock"] + str(item_id), stock)
# 活动信息
activity_key = f"seckill:activity:{item_id}"
pipe.hset(activity_key, mapping={
"start_time": start,
"end_time": end,
"total_stock": stock,
"status": "ready"
})
# 统计
stats_key = f"seckill:stats:{item_id}"
pipe.hset(stats_key, mapping={
"total_requests": 0,
"success_orders": 0,
"failed_orders": 0,
"start_time": time.time()
})
pipe.execute()
return True
# 使用示例
if __name__ == "__main__":
# 连接Redis
redis_client = redis.Redis(
host='localhost',
port=6379,
db=0,
decode_responses=True
)
# 初始化系统
system = EcommerceSeckillSystem(redis_client)
# 初始化秒杀活动
system.init_seckill_activity(
item_id="item_123",
stock_count=100,
start_time=time.time(),
end_time=time.time() + 3600
)
# 模拟用户秒杀
def simulate_user(user_id):
result = system.seckill(user_id, "item_123")
print(f"用户 {user_id}: {result}")
# 并发测试
with ThreadPoolExecutor(max_workers=10) as executor:
for i in range(200):
executor.submit(simulate_user, f"user_{i}")
# 查看统计
stats = system.get_activity_stats("item_123")
print(f"活动统计: {stats}")
总结与最佳实践
Redis在电商中的核心价值
性能提升
：毫秒级响应，支撑百万级并发
数据一致性
：原子操作保证业务准确性
架构解耦
：异步处理，提升系统吞吐量
实时计算
：用户行为实时分析与推荐
关键最佳实践
1. 数据结构选择
String
：缓存、计数器、分布式锁
Hash
：对象存储（购物车、用户信息）
List
：队列、消息通知
Set
：去重、共同关注
Sorted Set
：排行榜、优先级队列
2. 性能优化
# 使用pipeline批量操作
pipe = redis.pipeline()
for key in keys:
pipe.get(key)
results = pipe.execute()
# 使用Lua脚本保证原子性
lua_script = """
-- 复杂业务逻辑
"""
redis.eval(lua_script, ...)
# 合理使用过期时间
redis.setex(key, ttl, value)
3. 安全性
密码认证
：生产环境必须设置密码
网络隔离
：内网访问，限制公网暴露
命令禁用
：禁用危险命令（FLUSHALL、CONFIG等）
连接池
：避免连接泄漏
4. 监控与告警
# 监控关键指标
def monitor_redis():
info = redis.info()
# 内存使用率
memory_used = info['used_memory']
memory_total = info['maxmemory']
memory_ratio = memory_used / memory_total
if memory_ratio > 0.8:
send_alert(“内存使用率过高”)
# 命中率
hits = info[‘keyspace_hits’]
misses = info[‘keyspace_misses’]
hit_rate = hits / (hits + misses) if (hits + misses) > 0 else 0
if hit_rate < 0.5: send_alert("缓存命中率过低") 5. 高可用架构 主从复制 ：读写分离，提升性能 哨兵模式 ：自动故障转移 Redis Cluster ：数据分片，横向扩展 持久化策略 ：RDB+AOF混合持久化 未来发展趋势 Redis 6.0+ ：多线程IO、ACL、Client Cache Redis Modules ：RedisJSON、RedisSearch、RedisTimeSeries 云原生 ：Redis Enterprise、云托管服务 AI集成 ：向量搜索、机器学习特征存储 通过合理运用Redis，电商系统可以轻松应对高并发挑战，提供流畅的用户体验，同时保证数据的一致性和可靠性。掌握Redis的核心原理和最佳实践，是构建高性能电商系统的必备技能。