依赖和基础连接

pip install redis redis-py-cluster
import redis
from redis.cluster import RedisCluster
单机连接
r = redis.Redis(
    host='localhost', port=6379, db=0,
    password='your_password',
    decode_responses=True,
    socket_connect_timeout=5,
)
连接池（推荐）
pool = redis.ConnectionPool(host='localhost', port=6379, db=0, decode_responses=True)
r = redis.Redis(connection_pool=pool)
Cluster 模式
rc = RedisCluster(
    startup_nodes=[
        {'host': 'node1', 'port': 7001},
        {'host': 'node2', 'port': 7002},
        {'host': 'node3', 'port': 7003},
    ],
    decode_responses=True,
)

缓存层的简单封装

import json
import redis
from functools import wraps
class Cache:
    def __init__(self, redis_client, prefix='cache'):
        self.redis = redis_client
        self.prefix = prefix
def _key(self, key):
        return f'{self.prefix}:{key}'
def get(self, key):
        data = self.redis.get(self._key(key))
        if data is None:
            return None
        return json.loads(data)
def set(self, key, value, ttl=3600):
        self.redis.setex(self._key(key), ttl, json.dumps(value))
def delete(self, key):
        self.redis.delete(self._key(key))
def invalidate(self, pattern):
        keys = self.redis.keys(f'{self.prefix}:{pattern}')
        if keys:
            self.redis.delete(*keys)
def cached(ttl=3600, key_prefix=''):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            cache = getattr(wrapper, '_cache', None)
            if cache is None:
                wrapper._cache = Cache(redis.Redis(decode_responses=True))
cache_key = f'{key_prefix}:{func.__name__}:{str(args)}:{str(kwargs)}'
            cached_value = cache.get(cache_key)
            if cached_value is not None:
                return cached_value
result = func(*args, **kwargs)
            cache.set(cache_key, result, ttl)
            return result
        return wrapper
    return decorator
@cached(ttl=300, key_prefix='user')
def get_user(user_id):
    return {'id': user_id, 'name': f'User{user_id}'}

会话管理

import redis
import json
import uuid
from datetime import datetime

class SessionManager:
    def __init__(self, redis_client, prefix='session', ttl=86400):
        self.redis = redis_client
        self.prefix = prefix
        self.ttl = ttl
def create_session(self, user_id, extra_data=None):
        session_id = str(uuid.uuid4())
        session_key = f'{self.prefix}:{session_id}'
        session_data = {
            'user_id': user_id,
            'created_at': datetime.now().isoformat(),
            **(extra_data or {})
        }
        self.redis.hset(session_key, mapping={
            k: json.dumps(v) for k, v in session_data.items()
        })
        self.redis.expire(session_key, self.ttl)
        return session_id
def get_session(self, session_id):
        session_key = f'{self.prefix}:{session_id}'
        data = self.redis.hgetall(session_key)
        if not data:
            return None
        return {k: json.loads(v) for k, v in data.items()}
def destroy_session(self, session_id):
        self.redis.delete(f'{self.prefix}:{session_id}')

延迟任务队列

import redis
import json
import time

class DelayQueue:
    def __init__(self, redis_client, queue_name='delay:queue'):
        self.redis = redis_client
        self.queue_name = queue_name
        self.pending_name = f'{queue_name}:pending'
def push(self, task_id, delay_seconds, task_data):
        execute_at = time.time() + delay_seconds
        self.redis.zadd(self.queue_name, {task_id: execute_at})
        self.redis.hset(self.pending_name, task_id, json.dumps(task_data))
def pop(self, timeout=1):
        now = time.time()
        result = self.redis.zrangebyscore(self.queue_name, 0, now, start=0, num=1)
        if not result:
            time.sleep(timeout)
            return None
task_id = result[0]
        pipe = self.redis.pipeline()
        pipe.zrem(self.queue_name, task_id)
        pipe.hget(self.pending_name, task_id)
        pipe.hdel(self.pending_name, task_id)
        _, task_data, _ = pipe.execute()
if task_data:
            return task_id, json.loads(task_data)
        return None

排行榜

import redis

class Leaderboard:
    def __init__(self, redis_client, key='leaderboard'):
        self.redis = redis_client
        self.key = key
def add_score(self, member, score):
        self.redis.zadd(self.key, {member: score})
def increment(self, member, increment):
        self.redis.zincrby(self.key, increment, member)
def rank(self, member):
        rank = self.redis.zrevrank(self.key, member)
        return rank if rank is not None else None
def top(self, n=10):
        return self.redis.zrevrange(self.key, 0, n - 1, withscores=True)
def score(self, member):
        return self.redis.zscore(self.key, member)
lb = Leaderboard(r, 'article:views')
lb.add_score('article:1001', 1500)
lb.add_score('article:1002', 2000)
print(lb.top(3))