Redis 应用场景

缓存

Redis 最核心的应用场景：

import redis
import json

r = redis.Redis(host='localhost', port=6379)

def get_user(user_id):
    cache_key = f'user:{user_id}'
    cached = r.get(cache_key)

    if cached:
        return json.loads(cached)

    # 缓存未命中，查数据库
    user = query_database(user_id)

    # 写入缓存，设置5分钟过期
    r.setex(cache_key, 300, json.dumps(user))

    return user

缓存策略

策略 说明 优缺点
Cache Aside 应用维护缓存 简单，需处理一致性问题
Read/Write Through 缓存代理数据库 一致性好，实现复杂
Write Behind 异步写回DB 性能好，可能丢数据

分布式 Session

from flask import Flask, session
from flask_session import Session
import redis

app = Flask(__name__)
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] = redis.from_url('redis://localhost:6379')
Session(app)

@app.route('/login')
def login():
    session['user_id'] = 123
    session['username'] = 'Alice'
    return '已登录'

分布式锁

import uuid
import redis

r = redis.Redis()

class RedisLock:
    def __init__(self, key):
        self.key = f"lock:{key}"
        self.value = str(uuid.uuid4())

    def acquire(self, timeout=10):
        return r.set(self.key, self.value, nx=True, ex=timeout)

    def release(self):
        # Lua脚本保证原子性
        script = '''
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end
        '''
        r.eval(script, 1, self.key, self.value)

# 使用
lock = RedisLock("order:123")
if lock.acquire():
    try:
        process_order(123)
    finally:
        lock.release()

计数器

def incr_article_read(article_id):
    key = f'article:{article_id}:reads'
    return r.incr(key)

消息队列

# 生产者
def send_task(task_data):
    r.lpush('task:queue', json.dumps(task_data))

# 消费者
def worker():
    while True:
        task = r.brpop('task:queue', timeout=0)
        process_task(json.loads(task[1]))

排行榜

def add_score(player_id, score):
    r.zincrby('game:leaderboard', score, player_id)

def get_top(n=10):
    return r.zrevrange('game:leaderboard', 0, n-1, withscores=True)

最佳实践

  1. 设置合理的过期时间，避免内存占满
  2. 使用连接池管理连接
  3. 大 key 拆分，避免阻塞
  4. 监控慢查询日志
  5. 使用批量操作减少网络往返