08-redis实战应用

第八章：Redis实战应用

8.1 缓存系统实战

基础缓存

import redis
import json

# 创建Redis连接
r = redis.Redis(host='localhost', port=6379, db=0)

def get_user(user_id):
# 先从缓存获取
user = r.get(f'user:{user_id}')
if user:
return json.loads(user)

# 缓存不存在，从数据库获取
user = db.get_user(user_id)

# 写入缓存，设置过期时间
r.setex(f'user:{user_id}', 3600, json.dumps(user))

return user

8.2 分布式锁实战

简单分布式锁

import redis
import time

# 创建Redis连接
r = redis.Redis(host='localhost', port=6379, db=0)

def acquire_lock(lock_name, acquire_timeout=10, lock_timeout=10):
identifier = str(time.time())
end = time.time() + acquire_timeout

while time.time() < end:
if r.setnx(f'lock:{lock_name}', identifier):
r.expire(f'lock:{lock_name}', lock_timeout)
return identifier
time.sleep(0.001)

return False

def release_lock(lock_name, identifier):
pipe = r.pipeline(True)
while True:
try:
pipe.watch(f'lock:{lock_name}')
if pipe.get(f'lock:{lock_name}') == identifier:
pipe.multi()
pipe.delete(f'lock:{lock_name}')
pipe.execute()
return True
pipe.unwatch()
break
except redis.exceptions.WatchError:
pass

return False

8.3 排行榜实战

基础排行榜

import redis

# 创建Redis连接
r = redis.Redis(host='localhost', port=6379, db=0)

class Leaderboard:
def __init__(self, name):
self.name = f'leaderboard:{name}'

def add_score(self, user_id, score):
r.zadd(self.name, {user_id: score})

def get_rank(self, user_id):
rank = r.zrevrank(self.name, user_id)
return rank + 1 if rank is not None else None

def get_top_n(self, n):
return r.zrevrange(self.name, 0, n–1, withscores=True)

8.4 消息队列实战

简单消息队列

import redis
import json

# 创建Redis连接
r = redis.Redis(host='localhost', port=6379, db=0)

class SimpleQueue:
def __init__(self, name):
self.name = f'queue:{name}'

def push(self, message):
r.rpush(self.name, json.dumps(message))

def pop(self, timeout=0):
result = r.blpop(self.name, timeout=timeout)
if result:
return json.loads(result[1])
return None

8.5 限流系统实战

简单限流

import redis

# 创建Redis连接
r = redis.Redis(host='localhost', port=6379, db=0)

def is_allowed(user_id, limit=10, window=60):
key = f'limit:{user_id}'

script = '''
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local current = tonumber(redis.call('GET', key) or '0')

if current + 1 > limit then
return 0
else
redis.call('INCR', key)
if current == 0 then
redis.call('EXPIRE', key, window)
end
return 1
end
'''

return r.eval(script, 1, key, limit, window)

8.6 计数器系统实战

基础计数器

import redis

# 创建Redis连接
r = redis.Redis(host='localhost', port=6379, db=0)

def increment_counter(key):
return r.incr(key)

def get_counter(key):
return r.get(key)