django-redis 使用类实现和使用

在 Django 项目中使用 Redis

在 Django 项目中使用 Redis 可以通过多种方式实现，以下是完整的配置和使用指南：

1. 安装必要的包

首先安装 Django Redis 支持包：

pip install django-redis redis

2. 配置 Django 设置

在 settings.py 中添加 Redis 配置：

# settings.py

CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/1",  # 使用第1号数据库
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
            "PASSWORD": "your_password",  # 如果Redis有密码
            "SOCKET_CONNECT_TIMEOUT": 5,  # 连接超时时间（秒）
        },
        "KEY_PREFIX": "myapp_",  # 所有键的前缀
    }
}

# 可选：将session存储到Redis中
SESSION_ENGINE = "django.contrib.sessions.backends.cache"
SESSION_CACHE_ALIAS = "default"

3. 基本使用方法

方式一：使用 Django 缓存 API

from django.core.cache import cache

# 添加/设置键值
cache.set("user:1:name", "John Doe", timeout=3600)  # 1小时后过期
cache.set("user:1:email", "john@example.com")

# 获取值
username = cache.get("user:1:name")
email = cache.get("user:1:email")

print(f"Username: {username}, Email: {email}")

# 添加键（仅当键不存在时）
cache.add("user:1:age", 30)  # 只在键不存在时设置

# 设置多个键值
cache.set_many({"key1": "value1", "key2": "value2"})

# 获取多个键值
values = cache.get_many(["key1", "key2"])

# 删除键
cache.delete("user:1:name")

# 检查键是否存在
if cache.has_key("user:1:name"):
    print("键存在")

# 递增/递减值
cache.set("counter", 1)
cache.incr("counter")  # 2
cache.decr("counter")  # 1

方式二：使用原生 Redis 客户端

from django_redis import get_redis_connection

# 获取Redis连接
redis_client = get_redis_connection("default")

# 添加/设置键值
redis_client.set("user:2:name", "Jane Smith")
redis_client.setex("user:2:email", 3600, "jane@example.com")  # 带过期时间

# 获取值
username = redis_client.get("user:2:name")
if username:
    username = username.decode('utf-8')  # Redis返回字节串，需要解码

# 使用哈希表存储对象
redis_client.hset("user:3", "name", "Bob")
redis_client.hset("user:3", "email", "bob@example.com")
redis_client.expire("user:3", 3600)  # 设置过期时间

# 获取哈希表所有字段
user_data = redis_client.hgetall("user:3")
user_dict = {key.decode('utf-8'): value.decode('utf-8') for key, value in user_data.items()}

# 使用列表
redis_client.lpush("recent_users", "user1", "user2", "user3")
recent_users = redis_client.lrange("recent_users", 0, 2)  # 获取前3个元素

# 使用集合
redis_client.sadd("online_users", "user1", "user2")
online_users = redis_client.smembers("online_users")

4. 创建实用工具类

# utils/redis_utils.py
from django.core.cache import cache
from django_redis import get_redis_connection

class RedisHelper:
    @staticmethod
    def set_key(key, value, timeout=None):
        """设置键值"""
        return cache.set(key, value, timeout)
    
    @staticmethod
    def get_key(key, default=None):
        """获取键值"""
        return cache.get(key, default)
    
    @staticmethod
    def delete_key(key):
        """删除键"""
        return cache.delete(key)
    
    @staticmethod
    def get_client():
        """获取Redis原生客户端"""
        return get_redis_connection("default")
    
    @staticmethod
    def get_keys(pattern):
        """根据模式获取键列表"""
        client = get_redis_connection("default")
        return client.keys(pattern)
    
    @staticmethod
    def set_json(key, data, timeout=None):
        """存储JSON数据"""
        import json
        return cache.set(key, json.dumps(data), timeout)
    
    @staticmethod
    def get_json(key, default=None):
        """获取JSON数据"""
        import json
        value = cache.get(key)
        return json.loads(value) if value else default

5. 在视图中使用

# views.py
from django.http import JsonResponse
from django.views import View
from .utils.redis_utils import RedisHelper

class UserProfileView(View):
    def get(self, request, user_id):
        # 尝试从Redis获取用户数据
        user_data = RedisHelper.get_json(f"user:{user_id}:profile")
        
        if not user_data:
            # 如果Redis中没有，从数据库获取
            user = User.objects.get(id=user_id)
            user_data = {
                "name": user.name,
                "email": user.email,
                "last_login": user.last_login.isoformat()
            }
            # 存储到Redis，有效期1小时
            RedisHelper.set_json(f"user:{user_id}:profile", user_data, 3600)
        
        return JsonResponse(user_data)

6. 常用模式

缓存数据库查询结果

def get_popular_products():
    cache_key = "popular_products"
    products = cache.get(cache_key)
    
    if products is None:
        # 数据库查询
        products = list(Product.objects.filter(is_popular=True).values())
        # 缓存1小时
        cache.set(cache_key, products, 3600)
    
    return products

实现请求频率限制

from django.http import HttpResponseForbidden

def rate_limited_view(request):
    ip_address = request.META.get('REMOTE_ADDR')
    key = f"rate_limit:{ip_address}"
    
    # 获取当前计数
    requests = cache.get(key, 0)
    
    if requests >= 100:  # 限制100次请求
        return HttpResponseForbidden("Rate limit exceeded")
    
    # 增加计数
    cache.incr(key)
    # 设置过期时间（1小时）
    cache.expire(key, 3600)
    
    # 正常处理请求
    return HttpResponse("OK")

注意事项

1. 键命名规范：使用冒号分隔的命名空间（如 user:1:profile）
2. 序列化：复杂对象需要序列化（如使用JSON）
3. 内存管理：注意设置合理的过期时间，避免内存溢出
4. 错误处理：添加适当的异常处理
5. 连接池：Django-Redis默认使用连接池，无需手动管理连接

通过以上方式，你可以在Django项目中高效地使用Redis进行数据存储和缓存。