為了實現一個實時排行榜系統，我們可以使用Redis的有序集合（ZSet），其底層通常是使用跳躍表實現的。有序集合允許我們按照分數（score）對成員（member）進行排序，因此非常適合用來實現排行榜。本文首先介紹有序集合及其底層數據結構——跳表，然后使用Python和Redis結合，展示一個簡單的排行榜系統。
一、ZSet 概述
1.1 ZSet 介紹
實現一個排行榜，很多人可能首先想到的是使用MySQL的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 主要操作命令
ZADD key score member：將元素及其分數添加到有序集合中。ZINCRBY key increment member：為有序集合中的元素增加或減少分數。ZRANGE key start stop [WITHSCORES]：獲取有序集合中分數從小到大的排名在指定范圍內的成員。ZREVRANGE key start stop [WITHSCORES]：獲取有序集合中分數從大到小的排名在指定范圍內的成員。ZRANK key member：獲取成員在有序集合中的排名（從小到大的排名，排名從0開始）。ZREVRANK key member：獲取成員在有序集合中的排名（從大到小的排名，排名從0開始）。ZSCORE key member：獲取成員在有序集合中的分數。ZCARD key：獲取有序集合的基數，即成員數量。
二、使用 Redis 和 Python 實現實時排行榜
下面是一個使用Python的redis庫來操作ZSet并實現實時排行榜的示例。
2.1 安裝所需的庫
首先確保已經安裝redis庫：
pip install redis
2.2 初始化RedisLeaderboard類
接下來，我們實現一個RedisLeaderboard類來管理排行榜：
import redis
from flask import Flask, render_template
import sysapp = Flask(__name__)# Initialize Redis connection with error handling
try:r = redis.Redis(host='192.168.88.139',password='123456',port=6379,db=0,socket_connect_timeout=3, # 3 seconds timeoutdecode_responses=True # Automatically decode responses to UTF-8)# Test the connectionr.ping()print("成功連接Redis", file=sys.stderr)
except redis.ConnectionError as e:print(f"連接Redis失敗: {e}", file=sys.stderr)r = None # Set to None so we can check later@app.route('/')
def leaderboard():if r is None:return render_template('error.html',message="Redis server is not available"), 503try:top_10 = get_top_n(10)return render_template('leaderboard.html', leaderboard=top_10)except redis.RedisError as e:return render_template('error.html',message=f"Redis error: {str(e)}"), 500def get_top_n(n):try:top_n = r.zrevrange("game_leaderboard", 0, n - 1, withscores=True)leaderboard = []for rank, (user_id, score) in enumerate(top_n, start=1):leaderboard.append({"rank": rank,"user_id": user_id, # No need to decode with decode_responses=True"score": float(score)})return leaderboardexcept redis.RedisError as e:print(f"Redis operation failed: {e}", file=sys.stderr)raise # Re-raise the exception to be handled by the routeif __name__ == '__main__':app.run(debug=True)
2.3 案例數據
import redisr = redis.Redis(host='192.168.88.139', password='123456', port=6379, db=0)def add_score(user_id, score):r.zadd("game_leaderboard", {user_id: score})def update_score(user_id, score):r.zincrby("game_leaderboard", score, user_id)def get_top_n(n):top_n = r.zrevrange("game_leaderboard", 0, n - 1, withscores=True)leaderboard = []for rank, (user_id, score) in enumerate(top_n, start=1):leaderboard.append({"rank": rank,"user_id": user_id.decode("utf-8"),"score": score})return leaderboarddef get_user_rank_and_score(user_id):rank = r.zrevrank("game_leaderboard", user_id)if rank is not None:rank += 1score = r.zscore("game_leaderboard", user_id)return rank, scoreif __name__ == '__main__':# 添加初始得分add_score('user1', 100)add_score('user2', 150)add_score('user3', 50)# 更新得分（加分操作），如果用戶不存在，會將其得分初始化為該值update_score('user1', 30)update_score('user2', 20)update_score('user3', -10)# 獲取前2名的用戶top_2 = get_top_n(2)for entry in top_2:print(f"Rank {entry['rank']}: UserID: {entry['user_id']} with score {entry['score']}")# 獲取特定用戶的排名和得分rank, score = get_user_rank_and_score('user1')if rank is not None and score is not None:print(f"User user1 is ranked {rank} with a score of {score}.")else:print("User user1 is not found in the leaderboard.")
2.4 前端
需要創建一個templates文件夾，并在其中存放leaderboard.html文件：
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>Leaderboard</title><style>table {width: 100%;border-collapse: collapse;}th, td {border: 1px solid black;padding: 8px;text-align: left;}</style>
</head>
<body><h1>Leaderboard</h1><table><thead><tr><th>Rank</th><th>User ID</th><th>Score</th></tr></thead><tbody>{% for entry in leaderboard %}<tr><td>{{ entry.rank }}</td><td>{{ entry.user_id }}</td><td>{{ entry.score }}</td></tr>{% endfor %}</tbody></table>
</body>
</html>三、結論
Redis的有序集合(ZSet)由于其高效的插入、刪除、查詢及排序操作，是實現實時排行榜的理想選擇。跳表作為ZSet的底層數據結構之一，保證了這些操作的時間復雜度為O(log n)。結合Python的redis庫，可以快速實現一個功能強大、高效的實時排行榜系統。
這種排行榜實現方案非常適合用于在線游戲、社交平臺等各種應用場景。
