当前位置：首页 > news >正文

数据库技术：Redis缓存与分布式锁

news 2026/5/28 0:17:28

数据库技术：Redis缓存与分布式锁

大家好，我是欧阳瑞（Rich Own）。今天想和大家聊聊Redis缓存和分布式锁这个重要话题。作为一个全栈开发者，Redis是日常开发中不可或缺的工具。今天就来分享一下Redis缓存和分布式锁的实战经验。

Redis概述

Redis特性

特性	说明
内存存储	高性能读写
持久化	RDB和AOF
数据结构	字符串、哈希、列表、集合、有序集合
分布式	支持集群模式

应用场景

缓存 → 减轻数据库压力 会话存储 → 用户登录状态 分布式锁 → 保证并发安全 消息队列 → 简单的任务队列

Redis缓存

基本操作

import redis r = redis.Redis(host='localhost', port=6379, db=0) # 设置缓存 r.set('user:1', '{"name": "Alice", "age": 30}') # 获取缓存 data = r.get('user:1') print(data.decode()) # 设置过期时间 r.setex('session:abc123', 3600, 'user_id:1') # 删除缓存 r.delete('user:1')

缓存策略

def get_user(user_id): # 先查缓存 cache_key = f'user:{user_id}' cached = r.get(cache_key) if cached: return json.loads(cached.decode()) # 缓存不存在，查数据库 user = db.query('SELECT * FROM users WHERE id = %s', (user_id,)) if user: # 设置缓存，过期时间1小时 r.setex(cache_key, 3600, json.dumps(user)) return user

缓存更新

def update_user(user_id, data): # 更新数据库 db.query('UPDATE users SET name = %s WHERE id = %s', (data['name'], user_id)) # 失效缓存 r.delete(f'user:{user_id}')

分布式锁

基本实现

def acquire_lock(lock_key, timeout=10): """获取分布式锁""" identifier = str(uuid.uuid4()) end = time.time() + timeout while time.time() < end: if r.set(lock_key, identifier, nx=True, ex=timeout): return identifier time.sleep(0.001) return None def release_lock(lock_key, identifier): """释放分布式锁""" pipe = r.pipeline() pipe.watch(lock_key) if r.get(lock_key).decode() == identifier: pipe.multi() pipe.delete(lock_key) pipe.execute() return True pipe.unwatch() return False

使用示例

def process_order(order_id): lock_key = f'order:{order_id}:lock' identifier = acquire_lock(lock_key) if not identifier: raise Exception('操作过于频繁，请稍后重试') try: # 处理订单逻辑 order = get_order(order_id) update_inventory(order) send_notification(order) finally: release_lock(lock_key, identifier)

实战案例：限流

def rate_limit(user_id, limit=100, window=3600): """限流函数""" key = f'rate_limit:{user_id}' current = r.incr(key) if current == 1: r.expire(key, window) if current > limit: return False return True # 使用 if not rate_limit('user123'): raise Exception('请求过于频繁')

最佳实践

1. 缓存穿透处理

def get_user(user_id): cache_key = f'user:{user_id}' # 检查缓存 cached = r.get(cache_key) if cached == b'null': return None if cached: return json.loads(cached.decode()) # 查询数据库 user = db.query('SELECT * FROM users WHERE id = %s', (user_id,)) if user: r.setex(cache_key, 3600, json.dumps(user)) else: # 设置空值，防止缓存穿透 r.setex(cache_key, 60, 'null') return user

2. 缓存雪崩处理

# 使用随机过期时间，避免同时失效 def set_cache(key, value, base_ttl=3600): # 添加随机偏移，范围是基础TTL的10%-20% jitter = random.randint(int(base_ttl * 0.1), int(base_ttl * 0.2)) ttl = base_ttl + jitter r.setex(key, ttl, value)