1
import redis
2
import json
3

4
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
5

6
def get_user(user_id: int) -> dict | None:
7
    """Cache-Aside 读流程：先查缓存，miss 再查 DB"""
8
    cache_key = f"user:{user_id}"
9

10
    # 1. 查缓存
11
    cached = r.get(cache_key)
12
    if cached:
13
        return json.loads(cached)
14

15
    # 2. 未命中，查数据库
16
    user = db_query_user(user_id)  # 伪代码，实际为 ORM 查询
17
    if user is None:
18
        return None
19

20
    # 3. 回写缓存，设置过期时间避免脏数据永久驻留
21
    r.setex(cache_key, 3600, json.dumps(user))
22

23
    # 4. 返回结果
24
    return user
25

26
def db_query_user(user_id: int) -> dict | None:
27
    """模拟数据库查询"""
28
    # 实际项目中是 SELECT * FROM users WHERE id = ?
29
    return {"id": user_id, "name": "test", "email": "test@example.com"}

1
def update_user(user_id: int, new_name: str) -> None:
2
    """Cache-Aside 写流程：更新 DB，删缓存"""
3
    cache_key = f"user:{user_id}"
4

5
    # 1. 更新数据库
6
    db_update_user(user_id, new_name)
7

8
    # 2. 删除缓存（不是更新缓存）
9
    r.delete(cache_key)
10

11
def db_update_user(user_id: int, new_name: str) -> None:
12
    """模拟数据库更新"""
13
    # UPDATE users SET name = ? WHERE id = ?
14
    pass

1
def get_user_safe(user_id: int) -> dict | None:
2
    """缓存空值，防止穿透"""
3
    cache_key = f"user:{user_id}"
4

5
    cached = r.get(cache_key)
6
    if cached is not None:
7
        if cached == "NULL":
8
            # 命中空值缓存，直接返回 null
9
            return None
10
        return json.loads(cached)
11

12
    # 查数据库
13
    user = db_query_user(user_id)
14

15
    if user is None:
16
        # 数据库不存在，缓存空值，过期时间短（5 分钟）
17
        r.setex(cache_key, 300, "NULL")
18
        return None
19

20
    r.setex(cache_key, 3600, json.dumps(user))
21
    return user

1
def init_bloom_filter():
2
    """初始化布隆过滤器，加载所有合法 user_id"""
3
    # RedisBloom 模块提供 BF.ADD / BF.EXISTS 命令
4
    # 容量 100 万，误判率 0.1%
5
    r.execute_command('BF.RESERVE', 'bloom:user_ids', 0.001, 1000000)
6

7
    # 启动时加载所有 user_id
8
    all_ids = db_get_all_user_ids()  # SELECT id FROM users
9
    for uid in all_ids:
10
        r.execute_command('BF.ADD', 'bloom:user_ids', str(uid))
11

12
def get_user_with_bloom(user_id: int) -> dict | None:
13
    """布隆过滤器前置拦截，防止穿透"""
14
    # 先过布隆过滤器
15
    exists = r.execute_command('BF.EXISTS', 'bloom:user_ids', str(user_id))
16
    if exists == 0:
17
        # 布隆过滤器说不存在，一定不存在，直接返回
18
        return None
19

20
    # 布隆过滤器说可能存在，继续查缓存和 DB
21
    cache_key = f"user:{user_id}"
22
    cached = r.get(cache_key)
23
    if cached:
24
        return json.loads(cached)
25

26
    user = db_query_user(user_id)
27
    if user:
28
        r.setex(cache_key, 3600, json.dumps(user))
29
    return user

1
import time
2
import uuid
3

4
def get_hot_item(item_id: int) -> dict | None:
5
    """互斥锁防止击穿"""
6
    cache_key = f"item:{item_id}"
7
    lock_key = f"lock:item:{item_id}"
8

9
    # 1. 查缓存
10
    cached = r.get(cache_key)
11
    if cached:
12
        return json.loads(cached)
13

14
    # 2. 缓存 miss，尝试加互斥锁
15
    lock_value = str(uuid.uuid4())
16
    # SET NX PX：仅当 key 不存在时设置，并设置过期时间
17
    acquired = r.set(lock_key, lock_value, nx=True, px=10000)
18

19
    if acquired:
20
        try:
21
            # 双重检查：拿到锁后再查一次缓存
22
            cached = r.get(cache_key)
23
            if cached:
24
                return json.loads(cached)
25

26
            # 查数据库
27
            item = db_query_item(item_id)
28
            if item:
29
                r.setex(cache_key, 3600, json.dumps(item))
30
            return item
31
        finally:
32
            # 释放锁：用 Lua 保证判等与删除的原子性
33
            release_script = """
34
            if redis.call('get', KEYS[1]) == ARGV[1] then
35
                return redis.call('del', KEYS[1])
36
            else
37
                return 0
38
            end
39
            """
40
            r.eval(release_script, 1, lock_key, lock_value)
41
    else:
42
        # 未拿到锁，短暂等待后重试读缓存
43
        time.sleep(0.05)
44
        return get_hot_item(item_id)

1
import json
2
import time
3
import threading
4

5
def get_item_logical(item_id: int) -> dict | None:
6
    """逻辑过期：永远返回数据，异步刷新"""
7
    cache_key = f"item:logic:{item_id}"
8

9
    cached = r.get(cache_key)
10
    if not cached:
11
        # 冷启动，需要预热
12
        return None
13

14
    data = json.loads(cached)
15
    now = int(time.time())
16

17
    if data["expire_at"] > now:
18
        # 未过期，直接返回
19
        return data["value"]
20

21
    # 逻辑过期，尝试获取刷新锁
22
    lock_key = f"lock:refresh:{item_id}"
23
    acquired = r.set(lock_key, "1", nx=True, px=10000)
24

25
    if acquired:
26
        # 异步刷新，当前请求先返回旧值
27
        def refresh():
28
            try:
29
                item = db_query_item(item_id)
30
                new_data = {
31
                    "value": item,
32
                    "expire_at": int(time.time()) + 3600
33
                }
34
                r.set(cache_key, json.dumps(new_data))
35
            finally:
36
                r.delete(lock_key)
37

38
        threading.Thread(target=refresh, daemon=True).start()
39

40
    # 无论是否触发刷新，都返回旧数据
41
    return data["value"]

1
package main
2

3
import (
4
    "context"
5
    "encoding/json"
6
    "fmt"
7
    "time"
8

9
    "github.com/redis/go-redis/v9"
10
    "golang.org/x/sync/singleflight"
11
)
12

13
var (
14
    g   singleflight.Group
15
    ctx = context.Background()
16
)
17

18
func GetItem(rdb *redis.Client, itemID int) (map[string]any, error) {
19
    cacheKey := fmt.Sprintf("item:%d", itemID)
20

21
    // 1. 查缓存
22
    cached, err := rdb.Get(ctx, cacheKey).Result()
23
    if err == nil {
24
        var item map[string]any
25
        json.Unmarshal([]byte(cached), &item)
26
        return item, nil
27
    }
28

29
    // 2. 缓存 miss，用 singleflight 合并并发请求
30
    val, err, _ := g.Do(cacheKey, func() (any, error) {
31
        // 相同 key 的并发调用只执行一次这里的逻辑
32
        // 其他调用方阻塞等待结果
33
        item := dbQueryItem(itemID) // 伪代码
34
        if item != nil {
35
            data, _ := json.Marshal(item)
36
            rdb.Set(ctx, cacheKey, data, time.Hour)
37
        }
38
        return item, nil
39
    })
40

41
    if err != nil {
42
        return nil, err
43
    }
44
    return val.(map[string]any), nil
45
}

1
import random
2

3
def cache_item_with_jitter(item_id: int, value: dict) -> None:
4
    """设置随机过期时间，避免雪崩"""
5
    cache_key = f"item:{item_id}"
6
    base_ttl = 3600  # 基础 TTL：1 小时
7
    jitter = random.randint(0, 600)  # 随机偏移：0 到 10 分钟
8
    ttl = base_ttl + jitter
9
    r.setex(cache_key, ttl, json.dumps(value))
10

11
def batch_warmup(item_ids: list[int]) -> None:
12
    """批量预热缓存"""
13
    for item_id in item_ids:
14
        item = db_query_item(item_id)
15
        if item:
16
            # 每个 key 的 TTL 都不同，过期时间分散
17
            cache_item_with_jitter(item_id, item)

1
import random
2
import time
3
from functools import lru_cache
4

5
# 本地缓存（L1）：进程内，毫秒级访问
6
@lru_cache(maxsize=10000)
7
def get_item_local(item_id: int) -> tuple[dict | None, float]:
8
    """本地缓存，返回 (value, expire_time)"""
9
    return None, 0  # 实际由下面的函数填充
10

11
def get_item_multilevel(item_id: int) -> dict | None:
12
    """多级缓存：本地缓存 -> Redis -> 数据库"""
13
    cache_key = f"item:{item_id}"
14
    now = time.time()
15

16
    # L1：本地缓存
17
    local_val, local_expire = get_item_local(item_id)
18
    if local_val is not None and local_expire > now:
19
        return local_val
20

21
    # L2：Redis
22
    cached = r.get(cache_key)
23
    if cached:
24
        item = json.loads(cached)
25
        # 回写 L1，本地 TTL 短于 Redis TTL
26
        get_item_local.cache_clear()  # lru_cache 的简化回写
27
        return item
28

29
    # L3：数据库
30
    item = db_query_item(item_id)
31
    if item:
32
        r.setex(cache_key, 3600 + random.randint(0, 600), json.dumps(item))
33
    return item

1
import time
2
from collections import deque
3

4
class CircuitBreaker:
5
    """简单的熔断器：连续失败超阈值则熔断"""
6
    def __init__(self, threshold: int = 10, recovery_time: float = 60):
7
        self.threshold = threshold
8
        self.recovery_time = recovery_time
9
        self.failures = deque(maxlen=100)
10
        self.opened_at = 0
11

12
    def is_open(self) -> bool:
13
        """是否处于熔断状态"""
14
        if self.opened_at == 0:
15
            return False
16
        # 超过恢复时间，进入半开状态
17
        if time.time() - self.opened_at > self.recovery_time:
18
            self.opened_at = 0
19
            self.failures.clear()
20
            return False
21
        return True
22

23
    def record_failure(self):
24
        self.failures.append(time.time())
25
        if len(self.failures) >= self.threshold:
26
            self.opened_at = time.time()
27

28
breaker = CircuitBreaker(threshold=20, recovery_time=60)
29

30
def get_item_with_breaker(item_id: int) -> dict | None:
31
    """带熔断的查询：数据库挂了就降级"""
32
    if breaker.is_open():
33
        # 熔断中，返回降级数据
34
        return get_fallback_item(item_id)
35

36
    try:
37
        return get_item_multilevel(item_id)
38
    except Exception:
39
        breaker.record_failure()
40
        return get_fallback_item(item_id)
41

42
def get_fallback_item(item_id: int) -> dict | None:
43
    """降级：返回默认值或空"""
44
    return {"id": item_id, "name": "商品信息暂不可用"}

1
import time
2
import threading
3

4
def update_user_double_delete(user_id: int, new_name: str) -> None:
5
    """延迟双删：先删 -> 更新 DB -> 延迟再删"""
6
    cache_key = f"user:{user_id}"
7

8
    # 1. 第一次删除缓存
9
    r.delete(cache_key)
10

11
    # 2. 更新数据库
12
    db_update_user(user_id, new_name)
13

14
    # 3. 延迟第二次删除（异步，不阻塞主流程）
15
    delay_ms = 500  # 延迟 500ms
16
    def second_delete():
17
        time.sleep(delay_ms / 1000)
18
        r.delete(cache_key)
19

20
    threading.Thread(target=second_delete, daemon=True).start()

1
# 错误写法：非原子操作
2
stock = int(r.get("stock:1001") or 0)
3
if stock > 0:
4
    r.decr("stock:1001")

1
-- stock_deduct.lua
2
-- KEYS[1] = 库存 key，如 stock:1001
3
-- ARGV[1] = 扣减数量
4
-- ARGV[2] = 订单 ID（用于幂等）
5
-- 返回：1 成功，0 库存不足，-2 key 不存在
6

7
local stock_key = KEYS[1]
8
local deduct_amount = tonumber(ARGV[1])
9
local order_id = ARGV[2]
10

11
-- 检查是否已预扣过该订单（幂等性）
12
local dedup_key = "dedup:" .. order_id
13
if redis.call('EXISTS', dedup_key) == 1 then
14
    return 1  -- 已预扣过，直接返回成功
15
end
16

17
local current = redis.call('GET', stock_key)
18
if not current then
19
    return -2  -- 库存 key 不存在，需要初始化
20
end
21

22
current = tonumber(current)
23
if current < deduct_amount then
24
    return 0  -- 库存不足
25
end
26

27
-- 原子扣减
28
local remaining = redis.call('DECRBY', stock_key, deduct_amount)
29
if remaining < 0 then
30
    -- 扣减后为负，回滚并返回失败（并发竞态兜底）
31
    redis.call('INCRBY', stock_key, deduct_amount)
32
    return 0
33
end
34

35
-- 记录预扣订单，用于回滚和幂等
36
redis.call('SET', dedup_key, stock_key)
37
redis.call('EXPIRE', dedup_key, 1800)  -- 30 分钟过期
38

39
return 1

1
# 加载并执行 Lua 脚本
2
DEDUCT_SCRIPT = """
3
local stock_key = KEYS[1]
4
local deduct_amount = tonumber(ARGV[1])
5
local order_id = ARGV[2]
6

7
local dedup_key = "dedup:" .. order_id
8
if redis.call('EXISTS', dedup_key) == 1 then
9
    return 1
10
end
11

12
local current = redis.call('GET', stock_key)
13
if not current then
14
    return -2
15
end
16

17
current = tonumber(current)
18
if current < deduct_amount then
19
    return 0
20
end
21

22
local remaining = redis.call('DECRBY', stock_key, deduct_amount)
23
if remaining < 0 then
24
    redis.call('INCRBY', stock_key, deduct_amount)
25
    return 0
26
end
27

28
redis.call('SET', dedup_key, stock_key)
29
redis.call('EXPIRE', dedup_key, 1800)
30

31
return 1
32
"""
33

34
# 用 EVALSHA 执行，比 EVAL 省带宽（只传 SHA 不传脚本内容）
35
script_sha = r.script_load(DEDUCT_SCRIPT)
36

37
def deduct_stock(item_id: int, order_id: str, amount: int = 1) -> bool:
38
    """预扣库存"""
39
    stock_key = f"stock:{item_id}"
40
    result = r.evalsha(script_sha, 1, stock_key, str(amount), order_id)
41

42
    if result == 1:
43
        return True   # 预扣成功
44
    elif result == 0:
45
        return False  # 库存不足
46
    elif result == -2:
47
        # key 不存在，需要初始化库存
48
        return False  # 交由初始化流程处理
49
    return False

1
-- stock_rollback.lua
2
-- KEYS[1] = 库存 key
3
-- ARGV[1] = 回滚数量
4
-- ARGV[2] = 订单 ID
5

6
local stock_key = KEYS[1]
7
local rollback_amount = tonumber(ARGV[1])
8
local order_id = ARGV[2]
9

10
local dedup_key = "dedup:" .. order_id
11
-- 检查是否曾预扣过
12
if redis.call('EXISTS', dedup_key) == 0 then
13
    return 0  -- 没有预扣记录，无需回滚
14
end
15

16
-- 删除预扣记录（防止重复回滚）
17
redis.call('DEL', dedup_key)
18

19
-- 回滚库存
20
redis.call('INCRBY', stock_key, rollback_amount)
21
return 1

1
ROLLBACK_SCRIPT = """
2
local stock_key = KEYS[1]
3
local rollback_amount = tonumber(ARGV[1])
4
local order_id = ARGV[2]
5

6
local dedup_key = "dedup:" .. order_id
7
if redis.call('EXISTS', dedup_key) == 0 then
8
    return 0
9
end
10

11
redis.call('DEL', dedup_key)
12
redis.call('INCRBY', stock_key, rollback_amount)
13
return 1
14
"""
15

16
rollback_sha = r.script_load(ROLLBACK_SCRIPT)
17

18
def rollback_stock(item_id: int, order_id: str, amount: int = 1) -> bool:
19
    """回滚预扣库存"""
20
    stock_key = f"stock:{item_id}"
21
    result = r.evalsha(rollback_sha, 1, stock_key, str(amount), order_id)
22
    return result == 1

1
def place_order(user_id: int, item_id: int) -> dict:
2
    """下单流程：预扣库存 -> 创建订单"""
3
    order_id = generate_order_id()  # 伪代码
4

5
    # 1. Redis 预扣库存
6
    if not deduct_stock(item_id, order_id):
7
        return {"success": False, "msg": "库存不足或已售罄"}
8

9
    # 2. 创建订单（状态：待支付）
10
    order = create_order(order_id, user_id, item_id, status="pending")
11
    if not order:
12
        # 订单创建失败，回滚库存
13
        rollback_stock(item_id, order_id)
14
        return {"success": False, "msg": "订单创建失败"}
15

16
    # 3. 设置超时任务（30 分钟后自动取消）
17
    schedule_order_timeout(order_id, 1800)  # 伪代码
18
    return {"success": True, "order_id": order_id}
19

20
def confirm_payment(order_id: str) -> dict:
21
    """支付确认：DB 扣减库存，订单状态改为已支付"""
22
    order = get_order(order_id)  # 伪代码
23

24
    # DB 真正扣减库存
25
    affected = db_execute(
26
        "UPDATE stock SET count = count - 1 WHERE item_id = ? AND count > 0",
27
        order["item_id"]
28
    )
29
    if affected == 0:
30
        # DB 库存不足（Redis 预扣和 DB 不一致）
31
        rollback_stock(order["item_id"], order_id)
32
        return {"success": False, "msg": "库存不足"}
33

34
    update_order_status(order_id, "paid")
35
    return {"success": True}
36

37
def handle_timeout(order_id: str) -> None:
38
    """订单超时：回滚 Redis 库存，取消订单"""
39
    order = get_order(order_id)
40
    if order["status"] != "pending":
41
        return  # 已支付或已取消，不处理
42

43
    rollback_stock(order["item_id"], order_id)
44
    update_order_status(order_id, "cancelled")

1
def warmup_stock(item_id: int) -> None:
2
    """预热库存：从 DB 读取并写入 Redis"""
3
    db_stock = db_query_stock(item_id)  # SELECT count FROM stock WHERE item_id = ?
4
    stock_key = f"stock:{item_id}"
5
    # 用 SET 覆盖，确保是最新值
6
    r.set(stock_key, db_stock)
7

8
def sync_stock_to_db(item_id: int) -> None:
9
    """活动结束后：将 Redis 库存同步回 DB"""
10
    stock_key = f"stock:{item_id}"
11
    redis_stock = int(r.get(stock_key) or 0)
12
    db_execute(
13
        "UPDATE stock SET count = ? WHERE item_id = ?",
14
        redis_stock, item_id
15
    )
16
    # 同步后可删除 Redis 缓存
17
    r.delete(stock_key)

1
def update_score(user_id: int, score_delta: int) -> float:
2
    """更新积分：ZINCRBY 原子增减"""
3
    leaderboard_key = "leaderboard:daily"
4
    # ZINCRBY 原子性地增加分数，返回更新后的总分
5
    new_score = r.zincrby(leaderboard_key, score_delta, str(user_id))
6
    return new_score
7

8
def get_top_n(n: int = 10) -> list[dict]:
9
    """获取 Top N：ZREVRANGE 降序取前 N"""
10
    leaderboard_key = "leaderboard:daily"
11
    # ZREVRANGE key 0 n-1 WITHSCORES：分数从高到低
12
    results = r.zrevrange(leaderboard_key, 0, n - 1, withscores=True)
13
    return [
14
        {"user_id": int(member), "score": int(score)}
15
        for member, score in results
16
    ]
17

18
def get_user_rank(user_id: int) -> dict:
19
    """查个人排名和分数"""
20
    leaderboard_key = "leaderboard:daily"
21
    # ZREVRANK 返回降序排名（0 开始），ZSCORE 返回分数
22
    rank = r.zrevrank(leaderboard_key, str(user_id))
23
    score = r.zscore(leaderboard_key, str(user_id))
24
    if rank is None:
25
        return {"rank": -1, "score": 0}
26
    return {"rank": rank + 1, "score": int(score)}

1
def get_leaderboard_page(page: int, page_size: int = 20) -> dict:
2
    """分页获取排行榜"""
3
    leaderboard_key = "leaderboard:daily"
4
    start = (page - 1) * page_size
5
    end = start + page_size - 1
6

7
    # ZREVRANGE 按分数降序取指定范围
8
    results = r.zrevrange(leaderboard_key, start, end, withscores=True)
9
    total = r.zcard(leaderboard_key)  # 总人数
10

11
    return {
12
        "page": page,
13
        "page_size": page_size,
14
        "total": total,
15
        "items": [
16
            {"rank": start + i + 1, "user_id": int(member), "score": int(score)}
17
            for i, (member, score) in enumerate(results)
18
        ]
19
    }

1
import time
2

3
def get_neighbors(user_id: int, count: int = 2) -> list[dict]:
4
    """查看自己前后 count 名"""
5
    leaderboard_key = "leaderboard:daily"
6
    rank = r.zrevrank(leaderboard_key, str(user_id))
7
    if rank is None:
8
        return []
9

10
    # 取排名前后各 count 名
11
    start = max(0, rank - count)
12
    end = rank + count
13
    results = r.zrevrange(leaderboard_key, start, end, withscores=True)
14
    return [
15
        {"rank": start + i + 1, "user_id": int(member), "score": int(score)}
16
        for i, (member, score) in enumerate(results)
17
    ]
18

19
def cleanup_daily_leaderboard() -> None:
20
    """每日排行榜清理：用新 key 替换旧 key"""
21
    today = time.strftime("%Y%m%d")
22
    yesterday = (time.time() - 86400)
23
    yesterday_str = time.strftime("%Y%m%d", time.localtime(yesterday))
24

25
    old_key = f"leaderboard:{yesterday_str}"
26
    # 用 RENAME 原子替换，或直接删除历史数据
27
    r.delete(old_key)
28

29
    # 当天排行榜设置过期时间，自动清理
30
    today_key = f"leaderboard:{today}"
31
    r.expire(today_key, 86400 * 7)  # 7 天后自动删除

1
# 查看 Redis 命中统计
2
redis-cli INFO stats | grep -E "keyspace_hits|keyspace_misses"
3
# keyspace_hits:1523456
4
# keyspace_misses:82345
5

6
# 计算命中率
7
python3 -c "
8
hits=1523456; misses=82345
9
print(f'命中率: {hits/(hits+misses)*100:.2f}%')
10
"
11
# 命中率: 94.87%

1
# 扫描 bigkey，-i 0.1 表示每 100 个 key 休眠 0.1 秒，降低对线上影响
2
redis-cli --bigkeys -i 0.1
3

4
# 示例输出
5
# -------- summary -------
6
# Biggest string found so far: 'session:abc123' with 5.2 MB
7
# Biggest   list found so far: 'queue:tasks' with 128456 entries
8
# Biggest   hash found so far: 'user:profile:123' with 523 fields
9
# Biggest   zset found so far: 'rank:daily' with 234567 members

1
# 查看单个 key 的内存占用
2
redis-cli MEMORY USAGE rank:daily
3
# (integer) 15892384  # 约 15 MB
4

5
# 查看元素数量
6
redis-cli ZCARD rank:daily
7
# (integer) 234567
8

9
# 大 key 用 UNLINK 异步删除，避免阻塞
10
redis-cli UNLINK rank:daily

1
# 前提：必须配置 allkeys-lfu 或 volatile-lfu 策略
2
redis-cli CONFIG SET maxmemory-policy allkeys-lfu
3

4
# 扫描热 key
5
redis-cli --hotkeys
6

7
# 示例输出
8
# -------- sampling hotkeys --------
9
# -------- summary -------
10
# Sampled N keys in the keyspace!
11
# hot key found with counter: 6  keyname: "item:1001"
12
# hot key found with counter: 4  keyname: "user:session:xxx"

1
# 查看当前淘汰策略
2
redis-cli CONFIG GET maxmemory-policy
3
# 1) "maxmemory-policy"
4
# 2) "noeviction"
5

6
# 设置为 allkeys-lru（缓存场景最常用）
7
redis-cli CONFIG SET maxmemory-policy allkeys-lru
8

9
# 设置内存上限
10
redis-cli CONFIG SET maxmemory 4gb