List、Set、ZSet 精讲

List：双向链表与消息队列

Redis 的 List 是一个有序的字符串列表，支持从两端快速插入和弹出，天然适合队列（FIFO）和栈（LIFO）模式。

底层结构：quicklist

quicklist 兼顾了 linkedlist 的快速插删和 listpack 的内存紧凑性：两端操作 O(1)，中间查找 O(N)（不常用）。

List 命令速查表

消息队列模式（Python 实现）

import redis, json, time, threading

r = redis.Redis(host='localhost', decode_responses=True)

# 生产者：向队列推送任务
def producer(queue_name: str, task: dict) -> None:
    payload = json.dumps(task, ensure_ascii=False)
    r.rpush(queue_name, payload)
    print(f"[Producer] 推送任务: {task}")

# 消费者：阻塞式消费（推荐，避免轮询浪费 CPU）
def consumer(queue_name: str) -> None:
    print(f"[Consumer] 开始监听队列: {queue_name}")
    while True:
        # BLPOP：队列为空时阻塞等待，超时时间 0 表示永久等待
        result = r.blpop(queue_name, timeout=5)
        if result:
            _, payload = result
            task = json.loads(payload)
            print(f"[Consumer] 处理任务: {task}")
            process_task(task)  # 实际处理逻辑

def process_task(task: dict) -> None:
    time.sleep(0.1)  # 模拟处理
    print(f"[Worker] 完成: {task['id']}")

# 启动生产者和消费者
t = threading.Thread(target=consumer, args=('task:queue',), daemon=True)
t.start()

for i in range(5):
    producer('task:queue', {'id': i, 'type': 'email', 'to': f'user{i}@test.com'})

Set：无序集合与集合运算

Set 是无重复元素的无序集合，支持高效的集合运算（交集、并集、差集），适合标签系统、好友关系、抽奖等场景。

Set 命令速查表

# 社交网络：共同好友计算
SADD friends:alice bob charlie dave eve
SADD friends:bob  alice charlie frank

# 共同好友（交集）
SINTER friends:alice friends:bob
# 返回: {"charlie"}

# 推荐好友（bob 的好友中 alice 还没认识的）
SDIFF friends:bob friends:alice
# 返回: {"frank"}（排除了alice自己）

# 快速判断是否共同关注（交集基数，无需传输所有元素）
SINTERCARD 2 friends:alice friends:bob LIMIT 10
# 返回: 1

ZSet：有序集合与跳表

ZSet（Sorted Set）是 Redis 最独特的数据结构。每个元素有一个浮点数 score（分值），元素按 score 从小到大自动排序，同时支持按 score 范围查询和按 member 查询 score，复杂度均为 O(log N)。

底层实现：跳表（Skip List）

为何选择跳表而非红黑树？

• 范围查询效率：跳表底层是链表，ZRANGEBYSCORE 定位起点后直接顺序遍历，红黑树需要中序遍历
• 实现简单：跳表代码量约为红黑树的 1/3，更易维护和调试
• 内存局部性：跳表节点在插入顺序上连续，缓存命中率更高

ZSet 命令速查表

排行榜实现（完整代码）

import redis
from typing import List, Tuple

r = redis.Redis(host='localhost', decode_responses=True)
LEADERBOARD_KEY = "game:leaderboard"

def submit_score(player: str, score: float) -> int:
    """提交分数，GT 选项确保只有更高分才会更新"""
    r.zadd(LEADERBOARD_KEY, {player: score}, gt=True)
    # 获取当前排名（降序，+1 变成从1开始）
    rank = r.zrevrank(LEADERBOARD_KEY, player)
    return rank + 1 if rank is not None else -1

def get_top_n(n: int = 10) -> List[Tuple[str, float]]:
    """获取 Top N 排行榜"""
    # ZRANGE ... REV WITHSCORES：降序带分值（Redis 7.x 新语法）
    result = r.zrange(LEADERBOARD_KEY, 0, n - 1,
                       rev=True, withscores=True)
    return [(player, score) for player, score in result]

def get_player_rank(player: str) -> dict:
    """获取玩家的排名和分值"""
    with r.pipeline() as pipe:
        pipe.zrevrank(LEADERBOARD_KEY, player)
        pipe.zscore(LEADERBOARD_KEY, player)
        rank, score = pipe.execute()
    return {
        'player': player,
        'rank': (rank + 1) if rank is not None else None,
        'score': score
    }

def get_nearby_players(player: str, radius: int = 2) -> List[Tuple]:
    """获取玩家附近排名的玩家（如排名±2名）"""
    rank = r.zrevrank(LEADERBOARD_KEY, player)
    if rank is None: return []
    start = max(0, rank - radius)
    end = rank + radius
    return r.zrange(LEADERBOARD_KEY, start, end, rev=True, withscores=True)

# 测试
submit_score("alice", 9500)
submit_score("bob", 8800)
submit_score("charlie", 9200)
submit_score("dave", 7600)

print("Top 3:", get_top_n(3))
# [('alice', 9500.0), ('charlie', 9200.0), ('bob', 8800.0)]
print(get_player_rank("charlie"))
# {'player': 'charlie', 'rank': 2, 'score': 9200.0}