redis 实战练手-redis-action.zip

import json import threading import time import unittest import uuid from urllib.parse import urlparse, parse_qs from redis import Redis # region 登录 Cookie def check_token(conn: Redis, token: str) -> str: """ 通过令牌获取用户。 :param conn: Redis 连接 :param token: 用户令牌 :return: """ assert isinstance(conn, Redis) return conn.hget('login:', token) def update_token(conn: Redis, token: str, user: str, item: str = None) -> None: """ 记录用户最后一次浏览记录以及浏览时间。 :param conn: Redis 连接 :param token: 用户令牌 :param user: 用户 :param item: 浏览条目 :return: """ # 获取当前时间戳 timestamp = time.time() # 维持令牌与登录用户的映射 conn.hset('login:', token, user) # 更新令牌最后一次的设置时间 conn.zadd('recent:', token, timestamp) # 记录用户最近浏览记录，只保存 25 条记录 if item: conn.zadd('viewed:' + token, item, timestamp) conn.zremrangebyrank('viewed:' + token, 0, -26) # 为缓存页面，记录所有用户的浏览记录。 # 每一件商品浏览一次，对应的分数就减少 1 分， # 这样浏览次数越多的商品就越靠前。 conn.zincrby('viewed:', item, -1) QUIT = False LIMIT = 10000000 #令牌上限为 1000W，开发与测试时可调小 def clean_sessions(conn: Redis) -> None: """ 定期清理超过上限的会话相关记录。 定期清理会话，可以减轻内存压力。 清除会话时，一并清理用户登录信息，浏览记录和最近更新的令牌信息。 :param conn: Redis 连接 :return: """ while not QUIT: # 找出目前已有令牌的数量 size = conn.zcard('recent:') # 当目前已有令牌的数量没有超过上限时，休眠 1 秒 if size <= LIMIT: time.sleep(1) continue # 获取需要移除的令牌 ID，最多移除 100 个旧令牌 end_index = min(size - LIMIT, 100) tokens = conn.zrange('recent:', 0, end_index - 1) # 暂存要删除的令牌对应的浏览记录 session_keys = [] for token in tokens: session_keys.append('viewed:' + token) # 删除浏览记录 conn.delete(*session_keys) # 删除登录令牌 conn.hdel('login:', *tokens) # 删除令牌最近更新记录 conn.zrem('recent:', *tokens) # endregion # region 购物车 Cookie def add_to_cart(conn: Redis, session: str, item: str, count: int) -> None: """ 添加商品到购物车中，若商品数量小于等于 0，从购物车中删除该商品。 :param conn: Redis 连接 :param session: 登录用户会话 :param item: 商品名 :param count: 商品数量，设为 0 或 负数，将从购物车中移除该商品 :return: """ if count <= 0: # 商品数量不大于 0，从购物车中删除该商品 conn.hdel('cart:' + session, item) else: # 更新购物车中商品数量 conn.hset('cart:' + session, item, count) def clean_full_session(conn: Redis) -> None: """ 清除会话内容，在 clean_session 函数基础上再清除购物车信息。 :param conn: Redis 连接 :return: """ while not QUIT: size = conn.zcard('recent:') if size <= LIMIT: time.sleep(1) continue end_index = min(size - LIMIT, 100) sessions = conn.zrange('recent:', 0, end_index - 1) session_keys = [] for session in sessions: session_keys.append('viewed:' + session) session_keys.append('cart:' + session) conn.delete(*session_keys) conn.hdel('login:', *sessions) conn.zrem('recent:', *sessions) # endregion # region 缓存页面 def extract_item_id(request: str) -> str: """ 解析请求 URL，提取查询字符串中查询参数 item 的值。 若查询参数 item 不包含在请求 URL 中，返回 None。 :param request: 请求 URL :return: 返回查询参数 item 的值， 若 item 不存在，返回 None 若 item 有多个值，返回第一个值 """ # 解析 URL，分解成多个部分 parsed = urlparse(request) # 提取查询字符串部分 query = parse_qs(parsed.query) # 获取 item 查询参数的值 return (query.get('item') or [None])[0] def is_dynamic(request: str) -> bool: """ 如果请求 URL 的查询字符串中包括查询参数 '_'，表明该请求为动态请求，不可缓存。 :param request: 请求 URL :return: True - 动态资源请求 False - 静态资源请求 """ parsed = urlparse(request) query = parse_qs(parsed.query) return '_' in query def rescale_viewed(conn: Redis) -> None: """守护进程函数 优化页面缓存，低内存占用率。 当浏览记录超过 2W 条时，清空多余的不流行的商品页面缓存， 保留前 2W 条最流行的商品的缓存页面。 每次删除操作都将浏览次数减半，即降低分数的数值，但排名不变。 :param conn: Redis 连接 :return: """ while not QUIT: # 删除排名再 20000 名之后的浏览记录 conn.zremrangebyrank('viewed:', 20000, -1) # 将浏览次数降低为原来的一半 conn.zinterstore('viewed:', {'viewed:': .5}) # 5 分钟后重试 time.sleep(300) def can_cache(conn: Redis, request: str) -> bool: """ 判断请求是否可以被缓存。 缓存条件： 1. 静态请求 2. 商品页面的请求 3. 商品排名比较高 :param conn: Redis 连接 :param request: 请求 URL :return: True：可以缓存 False：不满足缓存条件 """ # 提取请求中的商品ID item_id = extract_item_id(request) # 检查是否为商品页面，是否可以静态缓存 if not item_id or is_dynamic(request): return False # 从已访问的商品列表中获取商品ID的排名 rank = conn.zrank('viewed:', item_id) # 检查商品是否达到缓存标准 return rank is not None and rank < 10000 def hash_request(request: str) -> str: """ 将请求 URL 字符串转换为散列。 :param request: 请求 URL :return: 返回散列形式的字符串 """ return str(hash(request)) def cache_request(conn: Redis, request: str, callback: any) -> str: # 如果不能缓存请求，直接调用回调函数 if not can_cache(conn, request): return callback(request) # 将请求转换成散列键 page_key = 'cache:' + hash_request(request) # 尝试查找缓存页面 content = conn.get(page_key) # 若没有缓存，生成页面并缓存 if not content: content = callback(request) conn.setex(page_key, content, 300) # 缓存 5 分钟 return content # endregion # region 缓存数据行 class Inventory(object): """ 商品库存类，这里主要模拟返回数据库的实际数据 """ def __init__(self, inv_id): self.id = inv_id @classmethod def get(cls, inv_id): return Inventory(inv_id) def to_dict(self): return {'id': self.id, 'data': '要缓存的数据', 'cached': time.time()} def schedule_row_cache(conn: Redis, row_id: str, delay: float) -> None: """ 对要缓存的数据行进行调度，为每行缓存数据设置缓存延迟时间（即该数据行下一次被缓存的时间间隔）。 这里的数据行指用户需要频繁读取关系数据库存储在硬盘里的数据。 :param conn: Redis 连接 :param row_id: 数据行 ID :param delay: 数据行缓存延迟时间 :