MonkeyCode与Redis实战：高并发体系的缓存计划完全指南

缓存是性能的杠杆

一个QPS 10000的体系，如果70%的哀求掷中缓存，数据库只必要遭受3000 QPS。缓存计划好，体系性能提升10倍不是标题。
这篇文章用MonkeyCode天生完备的Redis缓存方案，涵盖：缓存战略、分布式锁、耽误队列、排行榜、限流等实战场景。
给MonkeyCode的同一Prompt

1. 用Python + Redis实现高并发缓存系统，要求：
2. 1. 缓存策略：Cache-Aside、Write-Through、Write-Behind
3. 2. 缓存失效：TTL、LRU淘汰
4. 3. 分布式锁（RedLock）
5. 4. 延迟队列
6. 5. 排行榜（有序集合）
7. 6. 限流（滑动窗口、令牌桶）
8. 7. 消息发布订阅
9. 8. 热点数据探测
10. 9. 缓存异常降级处理
12. 需要完整可运行的代码，包含异常处理和监控埋点

复制代码

1. 缓存战略实现

1. # cache_strategies.py - MonkeyCode生成
2. import redis
3. import json
4. import hashlib
5. import time
6. from functools import wraps
7. from typing import Any, Callable, Optional
8. from dataclasses import dataclass
10. redis_client = redis.Redis(
11.     host='localhost',
12.     port=6379,
13.     db=0,
14.     decode_responses=True
15. )
17. # ============================================================
18. # Cache-Aside（旁路缓存）：最常用策略
19. # ============================================================
21. class CacheAside:
22.     """读多写少的场景首选"""
23.    
24.     def __init__(self, prefix: str = 'cache'):
25.         self.prefix = prefix
26.         self.default_ttl = 300  # 5分钟
27.    
28.     def _make_key(self, key: str) -> str:
29.         return f'{self.prefix}:{key}'
30.    
31.     def get(self, key: str, loader: Callable = None, ttl: int = None) -> Any:
32.         """获取缓存，未命中时加载并缓存"""
33.         redis_key = self._make_key(key)
34.         
35.         # 1. 读缓存
36.         cached = redis_client.get(redis_key)
37.         if cached:
38.             return json.loads(cached)
39.         
40.         # 2. 缓存未命中，调用loader加载
41.         if loader:
42.             data = loader()
43.             # 3. 写入缓存
44.             self.set(key, data, ttl or self.default_ttl)
45.             return data
46.         
47.         return None
48.    
49.     def set(self, key: str, value: Any, ttl: int = None):
50.         """写入缓存"""
51.         redis_key = self._make_key(key)
52.         redis_client.setex(
53.             redis_key,
54.             ttl or self.default_ttl,
55.             json.dumps(value)
56.         )
57.    
58.     def delete(self, key: str):
59.         """删除缓存"""
60.         redis_client.delete(self._make_key(key))
61.    
62.     def invalidate_pattern(self, pattern: str):
63.         """批量删除（匹配模式）"""
64.         keys = redis_client.keys(f'{self.prefix}:{pattern}')
65.         if keys:
66.             redis_client.delete(*keys)
68. # 使用示例
69. cache = CacheAside('user')
71. def get_user(user_id: int) -> dict:
72.     """获取用户（缓存优先）"""
73.     return cache.get(
74.         f'user:{user_id}',
75.         loader=lambda: load_user_from_db(user_id),
76.         ttl=600  # 用户信息缓存10分钟
77.     )
79. def update_user(user_id: int, data: dict):
80.     """更新用户时删除缓存"""
81.     update_user_to_db(user_id, data)
82.     cache.delete(f'user:{user_id}')
84. # ============================================================
85. # Write-Through（写穿透）：数据一致性强保证
86. # ============================================================
88. class WriteThrough:
89.     """写操作同时更新缓存和数据库"""
90.    
91.     def __init__(self, redis_client):
92.         self.redis = redis_client
93.    
94.     def write(self, key: str, value: Any, ttl: int = 3600):
95.         """同步写缓存和DB"""
96.         # 1. 写DB
97.         write_to_db(key, value)
98.         
99.         # 2. 写缓存
100.         self.redis.setex(key, ttl, json.dumps(value))
101.    
102.     def delete(self, key: str):
103.         """删除时同时清理缓存"""
104.         delete_from_db(key)
105.         self.redis.delete(key)
107. # ============================================================
108. # Write-Behind（写回）：写入性能最高
109. # ============================================================
111. class WriteBehind:
112.     """异步写回，批量写入减少IO"""
113.    
114.     def __init__(self, redis_client, queue_size: int = 1000):
115.         self.redis = redis_client
116.         self.queue_size = queue_size
117.         self.write_queue = []
118.    
119.     def write(self, key: str, value: Any):
120.         """先写缓存，异步批量写DB"""
121.         # 1. 写缓存
122.         self.redis.set(key, json.dumps(value))
123.         
124.         # 2. 加入写队列
125.         self.write_queue.append({'key': key, 'value': value})
126.         
127.         # 3. 达到阈值批量写DB
128.         if len(self.write_queue) >= self.queue_size:
129.             self.flush()
130.    
131.     def flush(self):
132.         """批量写DB"""
133.         if not self.write_queue:
134.             return
135.         
136.         batch_write_to_db(self.write_queue)
137.         self.write_queue.clear()

复制代码

2. 缓存装饰器

1. # cache_decorator.py - MonkeyCode生成
2. import redis
3. import json
4. import hashlib
5. import time
6. import functools
8. redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
10. def cached(key_prefix: str, ttl: int = 300, key_builder: Callable = None):
11.     """缓存装饰器"""
12.     def decorator(func):
13.         @functools.wraps(func)
14.         def wrapper(*args, **kwargs):
15.             # 构建缓存key
16.             if key_builder:
17.                 cache_key = key_builder(*args, **kwargs)
18.             else:
19.                 key_parts = [key_prefix, str(args), str(sorted(kwargs.items()))]
20.                 key_str = '|'.join(key_parts)
21.                 cache_key = f'{key_prefix}:{hashlib.md5(key_str.encode()).hexdigest()}'
22.             
23.             # 尝试获取缓存
24.             cached_value = redis_client.get(cache_key)
25.             if cached_value:
26.                 return json.loads(cached_value)
27.             
28.             # 执行原函数
29.             result = func(*args, **kwargs)
30.             
31.             # 写入缓存
32.             redis_client.setex(cache_key, ttl, json.dumps(result))
33.             
34.             return result
35.         return wrapper
36.     return decorator
38. def cache_invalidate(key_prefix: str, *args):
39.     """缓存失效装饰器"""
40.     def decorator(func):
41.         @functools.wraps(func)
42.         def wrapper(*args, **kwargs):
43.             result = func(*args, **kwargs)
44.             
45.             # 清理相关缓存
46.             pattern = f'{key_prefix}:*'
47.             keys = redis_client.keys(pattern)
48.             if keys:
49.                 redis_client.delete(*keys)
50.             
51.             return result
52.         return wrapper
53.     return decorator
55. # 使用
56. @cached('product', ttl=600)
57. def get_product(product_id: int) -> dict:
58.     """获取商品（缓存600秒）"""
59.     return query_product_from_db(product_id)
61. @cache_invalidate('product')
62. def update_product(product_id: int, data: dict):
63.     """更新商品时自动清理缓存"""
64.     return update_product_in_db(product_id, data)

复制代码

3. 分布式锁

1. # distributed_lock.py - MonkeyCode生成
2. import redis
3. import uuid
4. import time
5. import threading
6. from contextlib import contextmanager
7. from typing import Optional
9. class DistributedLock:
10.     """Redis分布式锁（RedLock简化版）"""
11.    
12.     def __init__(self, redis_client: redis.Redis, lock_name: str, timeout: int = 30):
13.         self.redis = redis_client
14.         self.lock_name = f'lock:{lock_name}'
15.         self.timeout = timeout
16.         self.token = str(uuid.uuid4())  # 锁持有者标识
17.    
18.     def acquire(self, blocking: bool = True, blocking_timeout: int = 10) -> bool:
19.         """
20.         获取锁
21.         blocking: 是否阻塞等待
22.         blocking_timeout: 阻塞超时时间（秒）
23.         """
24.         end_time = time.time() + blocking_timeout
25.         
26.         while True:
27.             # SET NX EX：原子操作
28.             acquired = self.redis.set(
29.                 self.lock_name,
30.                 self.token,
31.                 nx=True,  # 只有key不存在时设置
32.                 ex=self.timeout  # 自动过期
33.             )
34.             
35.             if acquired:
36.                 return True
37.             
38.             if not blocking:
39.                 return False
40.             
41.             if time.time() > end_time:
42.                 return False
43.             
44.             time.sleep(0.01)  # 避免CPU空转
45.    
46.     def release(self) -> bool:
47.         """
48.         释放锁（Lua脚本保证原子性）
49.         只有持有者才能释放自己的锁
50.         """
51.         lua_script = """
52.         if redis.call("get", KEYS[1]) == ARGV[1] then
53.             return redis.call("del", KEYS[1])
54.         else
55.             return 0
56.         end
57.         """
58.         result = self.redis.eval(lua_script, 1, self.lock_name, self.token)
59.         return result == 1
60.    
61.     def extend(self, additional_time: int = 30) -> bool:
62.         """延长锁的过期时间"""
63.         lua_script = """
64.         if redis.call("get", KEYS[1]) == ARGV[1] then
65.             return redis.call("expire", KEYS[1], ARGV[2])
66.         else
67.             return 0
68.         end
69.         """
70.         new_timeout = self.timeout + additional_time
71.         result = self.redis.eval(lua_script, 1, self.lock_name, self.token, new_timeout)
72.         return result == 1
73.    
74.     @contextmanager
75.     def lock(self, blocking: bool = True, blocking_timeout: int = 10):
76.         """上下文管理器方式使用锁"""
77.         acquired = self.acquire(blocking, blocking_timeout)
78.         if not acquired:
79.             raise RuntimeError(f'Could not acquire lock: {self.lock_name}')
80.         try:
81.             yield
82.         finally:
83.             self.release()
85. # 使用示例
86. lock = DistributedLock(redis_client, 'order:create:123', timeout=60)
88. # 方式1：手动获取/释放
89. if lock.acquire():
90.     try:
91.         # 临界区操作
92.         create_order(123)
93.     finally:
94.         lock.release()
96. # 方式2：上下文管理器（推荐）
97. with lock.lock():
98.     # 临界区操作
99.     create_order(123)

复制代码

4. 耽误队列

1. # delay_queue.py - MonkeyCode生成
2. import redis
3. import json
4. import time
5. import threading
6. from typing import Callable, Any, Optional
8. class DelayQueue:
9.     """Redis延迟队列（基于Sorted Set）"""
10.    
11.     def __init__(self, redis_client: redis.Redis, queue_name: str):
12.         self.redis = redis_client
13.         self.queue_name = f'delayq:{queue_name}'
14.         self.consumer_name = f'consumer:{queue_name}:{threading.current_thread().name}'
15.    
16.     def push(self, job_data: dict, delay_seconds: int):
17.         """添加延迟任务"""
18.         job_id = f'job:{time.time()}:{id(job_data)}'
19.         
20.         # 任务数据存在Hash里
21.         self.redis.hset(job_id, mapping={
22.             'data': json.dumps(job_data),
23.             'status': 'pending',
24.             'created_at': time.time()
25.         })
26.         
27.         # 任务ID和执行时间存入Sorted Set
28.         execute_at = time.time() + delay_seconds
29.         self.redis.zadd(self.queue_name, {job_id: execute_at})
30.         
31.         return job_id
32.    
33.     def pop(self, max_wait: int = 5) -> Optional[dict]:
34.         """
35.         阻塞获取可执行的任务
36.         """
37.         end_time = time.time() + max_wait
38.         
39.         while time.time() < end_time:
40.             # 获取1秒内过期的任务
41.             now = time.time()
42.             jobs = self.redis.zrangebyscore(
43.                 self.queue_name,
44.                 0,
45.                 now,
46.                 start=0,
47.                 num=1
48.             )
49.             
50.             if not jobs:
51.                 time.sleep(0.1)
52.                 continue
53.             
54.             job_id = jobs[0]
55.             
56.             # 原子性地将任务转移到处理中
57.             acquired = self.redis.zrem(self.queue_name, job_id)
58.             if not acquired:
59.                 # 被其他消费者抢走了
60.                 continue
61.             
62.             # 获取任务数据
63.             job_data = self.redis.hgetall(job_id)
64.             if not job_data:
65.                 continue
66.             
67.             # 更新状态为处理中
68.             self.redis.hset(job_id, 'status', 'processing')
69.             self.redis.hset(job_id, 'consumer', self.consumer_name)
70.             self.redis.hset(job_id, 'started_at', time.time())
71.             
72.             # 重新入队（如果超时未完成）
73.             self.redis.zadd(self.queue_name, {job_id: time.time() + 300})  # 5分钟超时
74.             
75.             return {
76.                 'job_id': job_id,
77.                 'data': json.loads(job_data['data']),
78.                 'created_at': float(job_data['created_at'])
79.             }
80.         
81.         return None
82.    
83.     def complete(self, job_id: str):
84.         """标记任务完成"""
85.         self.redis.hset(job_id, 'status', 'completed')
86.         self.redis.hset(job_id, 'completed_at', time.time())
87.    
88.     def fail(self, job_id: str, error: str):
89.         """标记任务失败"""
90.         self.redis.hset(job_id, 'status', 'failed')
91.         self.redis.hset(job_id, 'error', error)
92.         self.redis.hset(job_id, 'failed_at', time.time())
93.    
94.     def retry(self, job_id: str, delay_seconds: int = 60):
95.         """重试任务"""
96.         self.redis.hset(job_id, 'retry_count',
97.                        int(self.redis.hget(job_id, 'retry_count') or 0) + 1)
98.         self.redis.hset(job_id, 'status', 'pending')
99.         
100.         execute_at = time.time() + delay_seconds
101.         self.redis.zadd(self.queue_name, {job_id: execute_at})
103. # 使用示例
104. queue = DelayQueue(redis_client, 'order_timeout')
106. # 生产者：订单超时15分钟未支付则取消
107. def create_order(order_id: int):
108.     save_order_to_db(order_id)
109.     # 15分钟后检查订单支付状态
110.     queue.push({'order_id': order_id, 'action': 'cancel'}, delay_seconds=900)
112. # 消费者
113. def order_timeout_worker():
114.     while True:
115.         job = queue.pop(max_wait=5)
116.         if job:
117.             order_id = job['data']['order_id']
118.             if not is_order_paid(order_id):
119.                 cancel_order(order_id)
120.                 print(f'Order {order_id} cancelled due to timeout')
121.             queue.complete(job['job_id'])
123. # 启动消费者
124. worker_thread = threading.Thread(target=order_timeout_worker, daemon=True)
125. worker_thread.start()

复制代码

5. 排行榜

1. # leaderboard.py - MonkeyCode生成
2. import redis
3. from typing import List, Dict, Any
5. class Leaderboard:
6.     """Redis有序集合实现的排行榜"""
7.    
8.     def __init__(self, redis_client: redis.Redis, leaderboard_name: str):
9.         self.redis = redis_client
10.         self.key = f'leaderboard:{leaderboard_name}'
11.    
12.     def set_score(self, member: str, score: float):
13.         """设置/更新用户分数"""
14.         self.redis.zadd(self.key, {member: score})
15.    
16.     def increment(self, member: str, increment: float = 1):
17.         """增加分数"""
18.         self.redis.zincrby(self.key, increment, member)
19.    
20.     def get_rank(self, member: str, reverse: bool = False) -> int:
21.         """获取排名（0-based）"""
22.         if reverse:
23.             # 降序：第1名返回0
24.             return self.redis.zrevrank(self.key, member) or -1
25.         else:
26.             return self.redis.zrank(self.key, member) or -1
27.    
28.     def get_score(self, member: str) -> float:
29.         """获取分数"""
30.         return self.redis.zscore(self.key, member) or 0
31.    
32.     def get_top(self, top_n: int = 10, with_scores: bool = True) -> List[Dict[str, Any]]:
33.         """获取Top N"""
34.         if with_scores:
35.             results = self.redis.zrevrange(self.key, 0, top_n - 1, withscores=True)
36.             return [
37.                 {'rank': i + 1, 'member': member, 'score': score}
38.                 for i, (member, score) in enumerate(results)
39.             ]
40.         else:
41.             members = self.redis.zrevrange(self.key, 0, top_n - 1)
42.             return [{'rank': i + 1, 'member': m} for i, m in enumerate(members)]
43.    
44.     def get_around_me(self, member: str, count: int = 5) -> List[Dict[str, Any]]:
45.         """获取用户周围的排名"""
46.         rank = self.redis.zrevrank(self.key, member)
47.         if rank is None:
48.             return []
49.         
50.         start = max(0, rank - count)
51.         end = rank + count
52.         
53.         results = self.redis.zrevrange(self.key, start, end, withscores=True)
54.         
55.         return [
56.             {'rank': start + i + 1, 'member': m, 'score': s}
57.             for i, (m, s) in enumerate(results)
58.         ]
59.    
60.     def get_rank_by_score(self, min_score: float, max_score: float) -> int:
61.         """根据分数范围获取排名"""
62.         return self.redis.zcount(self.key, min_score, max_score)
64. # 使用示例
65. leaderboard = Leaderboard(redis_client, 'game_scores')
67. # 增加分数
68. leaderboard.increment('player_001', 100)
69. leaderboard.increment('player_002', 80)
70. leaderboard.increment('player_003', 120)
72. # 获取Top 10
73. top10 = leaderboard.get_top(10)
74. for p in top10:
75.     print(f"#{p['rank']} {p['member']}: {p['score']}")
77. # 查询特定玩家排名
78. player_rank = leaderboard.get_rank('player_001')
79. print(f"player_001 排名: #{player_rank + 1}")

复制代码

6. 限流

1. # rate_limiter.py - MonkeyCode生成
2. import redis
3. import time
4. from typing import Tuple
6. class SlidingWindowRateLimiter:
7.     """滑动窗口限流"""
8.    
9.     def __init__(self, redis_client: redis.Redis, key: str, max_requests: int, window_seconds: int):
10.         self.redis = redis_client
11.         self.key = f'ratelimit:{key}'
12.         self.max_requests = max_requests
13.         self.window_seconds = window_seconds
14.    
15.     def is_allowed(self) -> bool:
16.         """检查是否允许请求"""
17.         now = time.time()
18.         window_start = now - self.window_seconds
19.         
20.         # 移除窗口外的请求记录
21.         self.redis.zremrangebyscore(self.key, 0, window_start)
22.         
23.         # 检查当前请求数
24.         current_count = self.redis.zcard(self.key)
25.         
26.         if current_count >= self.max_requests:
27.             return False
28.         
29.         # 添加当前请求
30.         self.redis.zadd(self.key, {f'{now}': now})
31.         self.redis.expire(self.key, self.window_seconds)
32.         
33.         return True
34.    
35.     def get_remaining(self) -> int:
36.         """获取剩余可用次数"""
37.         now = time.time()
38.         window_start = now - self.window_seconds
39.         self.redis.zremrangebyscore(self.key, 0, window_start)
40.         return max(0, self.max_requests - self.redis.zcard(self.key))
42. class TokenBucketRateLimiter:
43.     """令牌桶限流"""
44.    
45.     def __init__(self, redis_client: redis.Redis, key: str,
46.                  capacity: int, refill_rate: float):
47.         self.redis = redis_client
48.         self.key = f'tokenbucket:{key}'
49.         self.capacity = capacity
50.         self.refill_rate = refill_rate  # 每秒补充令牌数
51.    
52.     def _refill(self) -> Tuple[float, float]:
53.         """补充令牌"""
54.         now = time.time()
55.         last_refill = float(self.redis.hget(self.key, 'last_refill') or now)
56.         
57.         # 计算应该补充的令牌数
58.         elapsed = now - last_refill
59.         tokens_to_add = elapsed * self.refill_rate
60.         
61.         current_tokens = float(self.redis.hget(self.key, 'tokens') or self.capacity)
62.         new_tokens = min(self.capacity, current_tokens + tokens_to_add)
63.         
64.         # 保存状态
65.         self.redis.hset(self.key, 'tokens', new_tokens)
66.         self.redis.hset(self.key, 'last_refill', now)
67.         self.redis.expire(self.key, 60)  # 1分钟过期
68.         
69.         return new_tokens
70.    
71.     def is_allowed(self, tokens_needed: int = 1) -> bool:
72.         """检查并消费令牌"""
73.         current_tokens = self._refill()
74.         
75.         if current_tokens >= tokens_needed:
76.             # 消费令牌
77.             new_tokens = current_tokens - tokens_needed
78.             self.redis.hset(self.key, 'tokens', new_tokens)
79.             return True
80.         
81.         return False
83. # 使用示例
84. api_limiter = SlidingWindowRateLimiter(redis_client, 'api:user:123', 60, 60)  # 每分钟60次
86. def api_handler(user_id: int):
87.     limiter = SlidingWindowRateLimiter(redis_client, f'api:user:{user_id}', 60, 60)
88.    
89.     if not limiter.is_allowed():
90.         raise Exception('请求过于频繁，请稍后再试')
91.    
92.     # 处理请求
93.     process_api_request(user_id)

复制代码

7. 热门数据探测

1. # hot_key_detection.py - MonkeyCode生成
2. import redis
3. import time
4. from collections import defaultdict
6. class HotKeyDetector:
7.     """热点Key探测（基于统计）"""
8.    
9.     def __init__(self, redis_client: redis.Redis, sample_rate: int = 100):
10.         self.redis = redis_client
11.         self.sample_rate = sample_rate  # 采样率（1/100）
12.         self.hot_keys = set()
13.    
14.     def record_access(self, key: str):
15.         """记录访问（采样）"""
16.         if time.time() % self.sample_rate == 0:
17.             self.redis.pfadd('access:keys', key)
18.    
19.     def get_hot_keys(self, threshold: int = 1000) -> set:
20.         """获取热点Key（HyperLogLog估算）"""
21.         # 导出统计
22.         keys = self.redis.pfcount('access:keys')
23.         if keys < threshold:
24.             return set()
25.         
26.         # 获取所有被访问过的Key
27.         # 实际应该用有序集合记录访问次数
28.         return self.hot_keys
30. # 基于访问频率的热Key缓存
31. class HotKeyCache:
32.     """热Key自动缓存"""
33.    
34.     def __init__(self, redis_client: redis.Redis):
35.         self.redis = redis_client
36.         self.prefix = 'hotcache:'
37.         self.hot_threshold = 1000  # 访问1000次以上
38.    
39.     def get(self, key: str, loader: Callable) -> Any:
40.         """获取数据，自动缓存热Key"""
41.         # 记录访问
42.         access_key = f'access:count:{key}'
43.         access_count = self.redis.incr(access_key)
44.         
45.         # 检查是否是热Key
46.         if access_count > self.hot_threshold and not self.redis.exists(f'{self.prefix}{key}'):
47.             # 预热缓存
48.             data = loader()
49.             self.redis.setex(f'{self.prefix}{key}', 3600, json.dumps(data))
50.         
51.         # 读缓存
52.         cached = self.redis.get(f'{self.prefix}{key}')
53.         if cached:
54.             return json.loads(cached)
55.         
56.         return loader()

复制代码

用Redis做缓存的关键是根据场景选战略：读多写少用Cache-Aside，写多读少用Write-Through，高并发抢锁用分布式锁，超时使命用耽误队列。MonkeyCode能帮你天生完备代码，但战略选择必要团毕业务特点。

免责声明：如果侵犯了您的权益，请联系站长及时删除侵权内容，谢谢合作！qidao123.com:ToB企服之家，中国第一个企服评测及软件市场,开放入驻,技术点评得现金.