Redis 实现排行榜的多种方案：ZSet、List 与自定义算法

各位观众老爷们，大家好！今天咱们来聊聊一个非常接地气的话题：排行榜！

排行榜这玩意儿，几乎在所有应用里都能看到它的身影，从游戏里的战力排名，到电商平台的销量榜单，再到社交应用的活跃度排行。别看它简单，要高效、准确地实现它，可不是一件容易的事。

今天，咱们就来扒一扒 Redis 实现排行榜的几种常见方案，看看它们各自的优缺点，以及在什么场景下更适合使用。

一、方案一：ZSet (有序集合)—— 榜上有名，按分排位

ZSet，也就是有序集合，是 Redis 提供的最适合实现排行榜的数据结构。它最大的特点就是：元素自带分数 (score)，Redis 会根据分数自动对元素进行排序。

1. ZSet 的基本原理

ZSet 实际上是一个跳跃表 (skiplist) 和哈希表 (hashtable) 的混合体。

• 跳跃表 (skiplist): 负责按照 score 进行排序，允许我们以 O(log N) 的时间复杂度进行范围查询，比如获取前 10 名的用户。
• 哈希表 (hashtable): 负责维护元素和 score 的映射关系，让我们能够以 O(1) 的时间复杂度通过元素找到对应的 score。

这种混合结构，既保证了排序的效率，又保证了查找的效率。

2. ZSet 的常用命令

• ZADD key score member: 添加元素到 ZSet，如果元素已经存在，则更新 score。
• ZRANGE key start stop [WITHSCORES]: 获取指定排名范围内的元素，可以携带分数。
• ZREVRANGE key start stop [WITHSCORES]: 逆序获取指定排名范围内的元素，可以携带分数。
• ZRANK key member: 获取元素的排名，排名从 0 开始。
• ZREVRANK key member: 逆序获取元素的排名，排名从 0 开始。
• ZSCORE key member: 获取元素的分数。
• ZINCRBY key increment member: 增加元素的分数。
• ZREM key member [member ...]: 移除一个或多个元素。
• ZCARD key: 获取ZSet的元素数量

3. ZSet 实现排行榜的代码示例 (Python + Redis)

import redis

# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 模拟一些用户数据
user_scores = {
    "user1": 1000,
    "user2": 1500,
    "user3": 800,
    "user4": 2000,
    "user5": 1200
}

# 将用户数据添加到 ZSet
for user, score in user_scores.items():
    r.zadd("leaderboard", {user: score})

# 获取前 3 名的用户
top_3 = r.zrevrange("leaderboard", 0, 2, withscores=True)
print("Top 3 users:")
for user, score in top_3:
    print(f"User: {user.decode()}, Score: {score}")

# 获取 user3 的排名
rank = r.zrevrank("leaderboard", "user3")
print(f"Rank of user3: {rank}")

# 增加 user1 的分数
r.zincrby("leaderboard", 500, "user1")

# 再次获取前 3 名的用户，看看 user1 的排名是否变化
top_3 = r.zrevrange("leaderboard", 0, 2, withscores=True)
print("Top 3 users after incrementing user1's score:")
for user, score in top_3:
    print(f"User: {user.decode()}, Score: {score}")

#移除user5
r.zrem("leaderboard", "user5")
print(f"Current size of leaderboard {r.zcard('leaderboard')}")

4. ZSet 的优点和缺点

• 优点:
  • 实现简单，Redis 提供了丰富的命令。
  • 性能高效，查询和更新的复杂度都是 O(log N)。
  • 自动排序，无需手动维护顺序。
• 缺点:
  • 占用内存相对较多，因为需要维护跳跃表和哈希表。
  • 当数据量非常大时，可能会成为性能瓶颈。

5. 适用场景

• 需要实时更新和查询排名的场景。
• 数据量不是特别大的场景（百万级别以下）。
• 对内存占用不敏感的场景。

二、方案二：List (列表) + 自定义排序—— 手动档，灵活定制

如果 ZSet 过于重量级，或者你需要更灵活的排序方式，那么可以考虑使用 List 结合自定义排序算法来实现排行榜。

1. List 的基本原理

List 是 Redis 提供的线性数据结构，可以存储一个有序的字符串列表。

2. List 的常用命令

• LPUSH key value [value ...]: 将一个或多个值插入到列表头部。
• RPUSH key value [value ...]: 将一个或多个值插入到列表尾部。
• LPOP key: 移除并返回列表头部的元素。
• RPOP key: 移除并返回列表尾部的元素。
• LRANGE key start stop: 获取指定范围内的元素。
• LINDEX key index: 获取指定索引的元素。
• LSET key index value: 设置指定索引的元素的值。
• LLEN key: 获取列表的长度。

3. List + 自定义排序实现排行榜的代码示例 (Python + Redis)

import redis

# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 模拟一些用户数据 (用户名:分数)
user_scores = {
    "user1": 1000,
    "user2": 1500,
    "user3": 800,
    "user4": 2000,
    "user5": 1200
}

def update_leaderboard(leaderboard_key, user, score):
    """
    更新排行榜，保持列表有序 (降序)
    """
    # 将用户和分数作为一个字符串存储，用冒号分隔
    user_score_str = f"{user}:{score}"

    # 获取当前排行榜
    leaderboard = r.lrange(leaderboard_key, 0, -1)
    leaderboard = [item.decode() for item in leaderboard]  # 将字节转换为字符串

    # 将新用户插入到合适的位置
    inserted = False
    for i, item in enumerate(leaderboard):
        existing_user, existing_score = item.split(":")
        existing_score = int(existing_score)
        if score > existing_score:
            leaderboard.insert(i, user_score_str)
            inserted = True
            break

    # 如果新用户分数最低，则添加到列表末尾
    if not inserted:
        leaderboard.append(user_score_str)

    # 保留前 N 名 (例如，前 10 名)
    leaderboard = leaderboard[:10]

    # 更新 Redis 列表
    r.delete(leaderboard_key)  # 清空旧列表
    if leaderboard:
        r.rpush(leaderboard_key, *leaderboard)  # 将新列表推入 Redis

# 初始化排行榜
leaderboard_key = "leaderboard_list"
for user, score in user_scores.items():
    update_leaderboard(leaderboard_key, user, score)

# 获取前 3 名的用户
top_3 = r.lrange(leaderboard_key, 0, 2)
print("Top 3 users:")
for item in top_3:
    user, score = item.decode().split(":")
    print(f"User: {user}, Score: {score}")

# 增加 user3 的分数
user3_new_score = 900
update_leaderboard(leaderboard_key, "user3", user3_new_score)

# 再次获取前 3 名的用户，看看 user3 的排名是否变化
top_3 = r.lrange(leaderboard_key, 0, 2)
print("Top 3 users after incrementing user3's score:")
for item in top_3:
    user, score = item.decode().split(":")
    print(f"User: {user}, Score: {score}")

4. List + 自定义排序的优点和缺点

• 优点:
  • 更加灵活，可以自定义排序算法，满足各种复杂的排序需求。
  • 内存占用相对较少，因为不需要维护额外的索引结构。
• 缺点:
  • 实现复杂，需要自己编写排序算法。
  • 性能较低，每次更新都需要重新排序整个列表，时间复杂度为 O(N)。

5. 适用场景

• 需要自定义排序算法的场景。
• 数据量比较小的场景（几百个元素以下）。
• 对性能要求不高的场景。

三、方案三：自定义算法 + Hash—— 自由发挥，性能至上

如果数据量巨大，而且对性能要求非常高，那么可以考虑使用自定义算法 + Hash 来实现排行榜。这种方案需要深入理解业务需求，并根据实际情况进行优化。

1. 基本思路

• Hash: 使用 Hash 存储用户的信息，包括用户名、分数等等。
• 自定义算法: 根据业务需求，设计一种高效的算法来维护排行榜的顺序。例如，可以使用分桶算法、二叉堆等等。

2. 分桶算法示例

分桶算法是一种常见的优化策略。它的基本思想是将用户按照分数范围划分到不同的桶中。

• 优点:
  • 可以减少排序的数据量，提高查询和更新的效率。
• 缺点:
  • 需要合理划分桶的大小，才能达到最佳的性能。
  • 实现复杂，需要考虑各种边界情况。

3. 代码示例 (Python + Redis，仅演示分桶思路)

import redis

# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 桶的大小
BUCKET_SIZE = 100

def get_bucket_key(score):
    """
    根据分数获取桶的 Key
    """
    bucket_index = score // BUCKET_SIZE
    return f"leaderboard:bucket:{bucket_index}"

def update_leaderboard_with_buckets(user, score):
    """
    使用分桶算法更新排行榜
    """
    # 获取用户之前的分数
    old_score = r.hget("user_scores", user)
    if old_score:
        old_score = int(old_score)

        # 从旧桶中移除用户
        old_bucket_key = get_bucket_key(old_score)
        r.srem(old_bucket_key, user)

    # 更新用户分数
    r.hset("user_scores", user, score)

    # 将用户添加到新桶中
    new_bucket_key = get_bucket_key(score)
    r.sadd(new_bucket_key, user)

def get_top_n_users_from_buckets(n):
    """
    从桶中获取前 N 名用户 (简化版，未考虑桶内排序)
    """
    top_users = []
    bucket_index = 9999 #假设最高分不会超过10000*BUCKET_SIZE
    while len(top_users) < n and bucket_index >= 0:
        bucket_key = f"leaderboard:bucket:{bucket_index}"
        users = r.smembers(bucket_key)
        users = [user.decode() for user in users]
        top_users.extend(users)
        bucket_index -= 1

    return top_users[:n]

# 模拟一些用户数据
user_scores = {
    "user1": 1000,
    "user2": 1500,
    "user3": 800,
    "user4": 2000,
    "user5": 1200
}

# 初始化排行榜
for user, score in user_scores.items():
    update_leaderboard_with_buckets(user, score)

# 获取前 3 名的用户
top_3 = get_top_n_users_from_buckets(3)
print("Top 3 users:")
print(top_3)

# 增加 user3 的分数
update_leaderboard_with_buckets("user3", 1800)

# 再次获取前 3 名的用户，看看 user3 的排名是否变化
top_3 = get_top_n_users_from_buckets(3)
print("Top 3 users after incrementing user3's score:")
print(top_3)

4. 自定义算法 + Hash 的优点和缺点

• 优点:
  • 性能最高，可以根据业务需求进行深度优化。
  • 可以处理海量数据。
• 缺点:
  • 实现最复杂，需要深入理解业务需求和数据结构。
  • 维护成本高。

5. 适用场景

• 数据量非常巨大，对性能要求非常高的场景。
• 需要根据业务需求进行深度定制的场景。

四、总结：选择合适的方案

方案	数据结构	优点	缺点	适用场景
ZSet (有序集合)	跳跃表 + 哈希表	实现简单，性能高效，自动排序	占用内存相对较多，数据量大时可能成为瓶颈	需要实时更新和查询排名，数据量不是特别大（百万级别以下），对内存占用不敏感的场景
List + 自定义排序	列表	更加灵活，可以自定义排序算法，内存占用相对较少	实现复杂，性能较低，每次更新都需要重新排序整个列表	需要自定义排序算法，数据量比较小（几百个元素以下），对性能要求不高的场景
自定义算法 + Hash	哈希表 + 自定义算法	性能最高，可以根据业务需求进行深度优化，可以处理海量数据	实现最复杂，需要深入理解业务需求和数据结构，维护成本高	数据量非常巨大，对性能要求非常高的场景，需要根据业务需求进行深度定制的场景

选择哪种方案，取决于你的具体需求。没有银弹，只有最合适的方案。

• 如果你的数据量不大，而且对性能要求不是特别高，那么 ZSet 是一个不错的选择。
• 如果你的排序逻辑比较复杂，需要自定义排序算法，那么 List + 自定义排序可能更适合你。
• 如果你的数据量非常巨大，而且对性能要求非常高，那么就需要考虑自定义算法 + Hash 了。

记住一点：永远不要过早优化。先选择最简单的方案，如果性能满足不了需求，再考虑更复杂的方案。

好了，今天的分享就到这里。希望大家有所收获！下次再见！