为了实现一个实时排行榜系统，我们可以使用Redis的有序集合（ZSet），其底层通常是使用跳跃表实现的。有序集合允许我们按照分数（score）对成员（member）进行排序，因此非常适合用来实现排行榜。本文首先介绍有序集合及其底层数据结构——跳表，然后使用Python和Redis结合，展示一个简单的排行榜系统。

一、ZSet 概述


1.1 ZSet 介绍

实现一个排行榜，很多人可能首先想到的是使用MySQL的

1. order by

复制代码

来排序。然而，当数据量达到百万级别时，使用数据库排序的代价是很大的。因此，Redis的有序集合（ZSet）成为了一个更好的选择。
ZSet（Sorted Set）的特点如下：

• 唯一性：集合内的元素（成员）是唯一的。
  
• 有序性：与普通Set的无序性不同，ZSet的成员是“有序的”，这种有序性是基于成员所关联的“分数”（score）进行排序的，分数是浮点类型。
  

1.2 Zset 底层原理

ZSet 是Redis中的一种复杂数据结构，它在Set的基础上增加了一个权重参数score，使得集合中的元素能按score进行有序排列。
ZSet的底层实现通常有两种数据结构：

• 当元素数量较少或元素长度较短时，采用压缩列表（ziplist）。
  
• 当元素数量达到一定量或者元素长度超过一定限制时，采用跳跃表（skiplist）。
  

跳表（skiplist）具有多层链表结构，查询、插入和删除操作的平均时间复杂度均为O(log n)。

1.3 ZSet 主要操作命令

• 1. ZADD key score member

  复制代码
  ：将元素及其分数添加到有序集合中。
  
• 1. ZINCRBY key increment member

  复制代码
  ：为有序集合中的元素增加或减少分数。
  
• 1. ZRANGE key start stop [WITHSCORES]

  复制代码
  ：获取有序集合中分数从小到大的排名在指定范围内的成员。
  
• 1. ZREVRANGE key start stop [WITHSCORES]

  复制代码
  ：获取有序集合中分数从大到小的排名在指定范围内的成员。
  
• 1. ZRANK key member

  复制代码
  ：获取成员在有序集合中的排名（从小到大的排名，排名从0开始）。
  
• 1. ZREVRANK key member

  复制代码
  ：获取成员在有序集合中的排名（从大到小的排名，排名从0开始）。
  
• 1. ZSCORE key member

  复制代码
  ：获取成员在有序集合中的分数。
  
• 1. ZCARD key

  复制代码
  ：获取有序集合的基数，即成员数量。
  

二、使用 Redis 和 Python 实现实时排行榜

下面是一个使用Python的

1. redis

复制代码

库来操作ZSet并实现实时排行榜的示例。

2.1 安装所需的库

首先确保已经安装

1. redis

复制代码

库：

1. pip install redis

复制代码

2.2 初始化RedisLeaderboard类

接下来，我们实现一个

1. RedisLeaderboard

复制代码

类来管理排行榜：

1. import redis
   
2. from flask import Flask, render_template
   
3. import sys
   
4. 
   
5. app = Flask(__name__)
   
6. 
   
7. # Initialize Redis connection with error handling
   
8. try:
   
9.     r = redis.Redis(
   
10.         host='192.168.88.139',
    
11.         password='123456',
    
12.         port=6379,
    
13.         db=0,
    
14.         socket_connect_timeout=3,  # 3 seconds timeout
    
15.         decode_responses=True  # Automatically decode responses to UTF-8
    
16.     )
    
17.     # Test the connection
    
18.     r.ping()
    
19.     print("成功连接Redis", file=sys.stderr)
    
20. except redis.ConnectionError as e:
    
21.     print(f"连接Redis失败: {e}", file=sys.stderr)
    
22.     r = None  # Set to None so we can check later
    
23. 
    
24. 
    
25. @app.route('/')
    
26. def leaderboard():
    
27.     if r is None:
    
28.         return render_template('error.html',
    
29.                                message="Redis server is not available"), 503
    
30. 
    
31.     try:
    
32.         top_10 = get_top_n(10)
    
33.         return render_template('leaderboard.html', leaderboard=top_10)
    
34.     except redis.RedisError as e:
    
35.         return render_template('error.html',
    
36.                                message=f"Redis error: {str(e)}"), 500
    
37. 
    
38. 
    
39. def get_top_n(n):
    
40.     try:
    
41.         top_n = r.zrevrange("game_leaderboard", 0, n - 1, withscores=True)
    
42.         leaderboard = []
    
43.         for rank, (user_id, score) in enumerate(top_n, start=1):
    
44.             leaderboard.append({
    
45.                 "rank": rank,
    
46.                 "user_id": user_id,  # No need to decode with decode_responses=True
    
47.                 "score": float(score)
    
48.             })
    
49.         return leaderboard
    
50.     except redis.RedisError as e:
    
51.         print(f"Redis operation failed: {e}", file=sys.stderr)
    
52.         raise  # Re-raise the exception to be handled by the route
    
53. 
    
54. 
    
55. if __name__ == '__main__':
    
56.     app.run(debug=True)

复制代码

2.3 案例数据

1. import redis
   
2. 
   
3. r = redis.Redis(host='192.168.88.139', password='123456', port=6379, db=0)
   
4. 
   
5. 
   
6. def add_score(user_id, score):
   
7.     r.zadd("game_leaderboard", {user_id: score})
   
8. 
   
9. 
   
10. def update_score(user_id, score):
    
11.     r.zincrby("game_leaderboard", score, user_id)
    
12. 
    
13. 
    
14. def get_top_n(n):
    
15.     top_n = r.zrevrange("game_leaderboard", 0, n - 1, withscores=True)
    
16.     leaderboard = []
    
17.     for rank, (user_id, score) in enumerate(top_n, start=1):
    
18.         leaderboard.append({
    
19.             "rank": rank,
    
20.             "user_id": user_id.decode("utf-8"),
    
21.             "score": score
    
22.         })
    
23.     return leaderboard
    
24. 
    
25. 
    
26. def get_user_rank_and_score(user_id):
    
27.     rank = r.zrevrank("game_leaderboard", user_id)
    
28.     if rank is not None:
    
29.         rank += 1
    
30.     score = r.zscore("game_leaderboard", user_id)
    
31.     return rank, score
    
32. 
    
33. 
    
34. if __name__ == '__main__':
    
35.     # 添加初始得分
    
36.     add_score('user1', 100)
    
37.     add_score('user2', 150)
    
38.     add_score('user3', 50)
    
39. 
    
40.     # 更新得分（加分操作），如果用户不存在，会将其得分初始化为该值
    
41.     update_score('user1', 30)
    
42.     update_score('user2', 20)
    
43.     update_score('user3', -10)
    
44. 
    
45.     # 获取前2名的用户
    
46.     top_2 = get_top_n(2)
    
47.     for entry in top_2:
    
48.         print(f"Rank {entry['rank']}: UserID: {entry['user_id']} with score {entry['score']}")
    
49. 
    
50.     # 获取特定用户的排名和得分
    
51.     rank, score = get_user_rank_and_score('user1')
    
52.     if rank is not None and score is not None:
    
53.         print(f"User user1 is ranked {rank} with a score of {score}.")
    
54.     else:
    
55.         print("User user1 is not found in the leaderboard.")

复制代码

2.4 前端

需要创建一个templates文件夹，并在其中存放leaderboard.html文件：

1. <!DOCTYPE html>
   
2. <html lang="en">
   
3. <head>
   
4.     <meta charset="UTF-8">
   
5.     <meta name="viewport" content="width=device-width, initial-scale=1.0">
   
6.     <title>Leaderboard</title>
   
7.     <style>
   
8.         table {
   
9.             width: 100%;
   
10.             border-collapse: collapse;
    
11.         }
    
12.         th, td {
    
13.             border: 1px solid black;
    
14.             padding: 8px;
    
15.             text-align: left;
    
16.         }
    
17.     </style>
    
18. </head>
    
19. <body>
    
20.     <h1>Leaderboard</h1>
    
21.     <table>
    
22.         <thead>
    
23.             <tr>
    
24.                 <th>Rank</th>
    
25.                 <th>User ID</th>
    
26.                 <th>Score</th>
    
27.             </tr>
    
28.         </thead>
    
29.         <tbody>
    
30.             {% for entry in leaderboard %}
    
31.             <tr>
    
32.                 <td>{{ entry.rank }}</td>
    
33.                 <td>{{ entry.user_id }}</td>
    
34.                 <td>{{ entry.score }}</td>
    
35.             </tr>
    
36.             {% endfor %}
    
37.         </tbody>
    
38.     </table>
    
39. </body>
    
40. </html>

复制代码

三、结论

Redis的有序集合(ZSet)由于其高效的插入、删除、查询及排序操作，是实现实时排行榜的理想选择。跳表作为ZSet的底层数据结构之一，保证了这些操作的时间复杂度为O(log n)。结合Python的

1. redis

复制代码

库，可以快速实现一个功能强大、高效的实时排行榜系统。
这种排行榜实现方案非常适合用于在线游戏、社交平台等各种应用场景。
到此这篇关于使用Redis实现实时排行榜的示例的文章就介绍到这了,更多相关Redis 实时排行榜内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

来源：https://www.jb51.net/database/339991pl9.htm
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！