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

django-redis 使用类实现和使用

news 2025/8/28 11:38:50

在 Django 项目中使用 Redis

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

1. 安装必要的包

首先安装 Django Redis 支持包：

pip install django-redis redis

2. 配置 Django 设置

在 settings.py 中添加 Redis 配置：

# settings.pyCACHES = {"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_connectionclass RedisHelper:@staticmethoddef set_key(key, value, timeout=None):"""设置键值"""return cache.set(key, value, timeout)@staticmethoddef get_key(key, default=None):"""获取键值"""return cache.get(key, default)@staticmethoddef delete_key(key):"""删除键"""return cache.delete(key)@staticmethoddef get_client():"""获取Redis原生客户端"""return get_redis_connection("default")@staticmethoddef get_keys(pattern):"""根据模式获取键列表"""client = get_redis_connection("default")return client.keys(pattern)@staticmethoddef set_json(key, data, timeout=None):"""存储JSON数据"""import jsonreturn cache.set(key, json.dumps(data), timeout)@staticmethoddef get_json(key, default=None):"""获取JSON数据"""import jsonvalue = 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 RedisHelperclass 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 HttpResponseForbiddendef 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")