快捷導(dǎo)航

使用Redis實(shí)現(xiàn)實(shí)時(shí)排行榜的示例

更新時(shí)間：2025年04月21日 08:33:45 作者：庫庫林_沙琪馬

為了實(shí)現(xiàn)一個(gè)實(shí)時(shí)排行榜系統(tǒng),我們可以使用Redis的有序集合,本文主要介紹了使用Redis實(shí)現(xiàn)實(shí)時(shí)排行榜的示例,具有一定的參考價(jià)值,感興趣的可以了解一下

為了實(shí)現(xiàn)一個(gè)實(shí)時(shí)排行榜系統(tǒng)，我們可以使用Redis的有序集合（ZSet），其底層通常是使用跳躍表實(shí)現(xiàn)的。有序集合允許我們按照分?jǐn)?shù)（score）對成員（member）進(jìn)行排序，因此非常適合用來實(shí)現(xiàn)排行榜。本文首先介紹有序集合及其底層數(shù)據(jù)結(jié)構(gòu)——跳表，然后使用Python和Redis結(jié)合，展示一個(gè)簡單的排行榜系統(tǒng)。

一、ZSet 概述

1.1 ZSet 介紹

實(shí)現(xiàn)一個(gè)排行榜，很多人可能首先想到的是使用MySQL的order by來排序。然而，當(dāng)數(shù)據(jù)量達(dá)到百萬級別時(shí)，使用數(shù)據(jù)庫排序的代價(jià)是很大的。因此，Redis的有序集合（ZSet）成為了一個(gè)更好的選擇。

ZSet（Sorted Set）的特點(diǎn)如下：

唯一性：集合內(nèi)的元素（成員）是唯一的。
有序性：與普通Set的無序性不同，ZSet的成員是“有序的”，這種有序性是基于成員所關(guān)聯(lián)的“分?jǐn)?shù)”（score）進(jìn)行排序的，分?jǐn)?shù)是浮點(diǎn)類型。

1.2 Zset 底層原理

ZSet 是Redis中的一種復(fù)雜數(shù)據(jù)結(jié)構(gòu)，它在Set的基礎(chǔ)上增加了一個(gè)權(quán)重參數(shù)score，使得集合中的元素能按score進(jìn)行有序排列。

ZSet的底層實(shí)現(xiàn)通常有兩種數(shù)據(jù)結(jié)構(gòu)：

當(dāng)元素?cái)?shù)量較少或元素長度較短時(shí)，采用壓縮列表（ziplist）。
當(dāng)元素?cái)?shù)量達(dá)到一定量或者元素長度超過一定限制時(shí)，采用跳躍表（skiplist）。

跳表（skiplist）具有多層鏈表結(jié)構(gòu)，查詢、插入和刪除操作的平均時(shí)間復(fù)雜度均為O(log n)。

1.3 ZSet 主要操作命令

ZADD key score member：將元素及其分?jǐn)?shù)添加到有序集合中。
ZINCRBY key increment member：為有序集合中的元素增加或減少分?jǐn)?shù)。
ZRANGE key start stop [WITHSCORES]：獲取有序集合中分?jǐn)?shù)從小到大的排名在指定范圍內(nèi)的成員。
ZREVRANGE key start stop [WITHSCORES]：獲取有序集合中分?jǐn)?shù)從大到小的排名在指定范圍內(nèi)的成員。
ZRANK key member：獲取成員在有序集合中的排名（從小到大的排名，排名從0開始）。
ZREVRANK key member：獲取成員在有序集合中的排名（從大到小的排名，排名從0開始）。
ZSCORE key member：獲取成員在有序集合中的分?jǐn)?shù)。
ZCARD key：獲取有序集合的基數(shù)，即成員數(shù)量。

二、使用 Redis 和 Python 實(shí)現(xiàn)實(shí)時(shí)排行榜

下面是一個(gè)使用Python的redis庫來操作ZSet并實(shí)現(xiàn)實(shí)時(shí)排行榜的示例。

2.1 安裝所需的庫

首先確保已經(jīng)安裝redis庫：

pip install redis

2.2 初始化RedisLeaderboard類

接下來，我們實(shí)現(xiàn)一個(gè)RedisLeaderboard類來管理排行榜：

import redis
from flask import Flask, render_template
import sys

app = 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 timeout
        decode_responses=True  # Automatically decode responses to UTF-8
    )
    # Test the connection
    r.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"), 503

    try:
        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)}"), 500


def 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 leaderboard
    except redis.RedisError as e:
        print(f"Redis operation failed: {e}", file=sys.stderr)
        raise  # Re-raise the exception to be handled by the route


if __name__ == '__main__':
    app.run(debug=True)

在這里插入圖片描述

2.3 案例數(shù)據(jù)

import redis

r = 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 leaderboard


def get_user_rank_and_score(user_id):
    rank = r.zrevrank("game_leaderboard", user_id)
    if rank is not None:
        rank += 1
    score = r.zscore("game_leaderboard", user_id)
    return rank, score


if __name__ == '__main__':
    # 添加初始得分
    add_score('user1', 100)
    add_score('user2', 150)
    add_score('user3', 50)

    # 更新得分（加分操作），如果用戶不存在，會(huì)將其得分初始化為該值
    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 前端

需要?jiǎng)?chuàng)建一個(gè)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>

三、結(jié)論

Redis的有序集合(ZSet)由于其高效的插入、刪除、查詢及排序操作，是實(shí)現(xiàn)實(shí)時(shí)排行榜的理想選擇。跳表作為ZSet的底層數(shù)據(jù)結(jié)構(gòu)之一，保證了這些操作的時(shí)間復(fù)雜度為O(log n)。結(jié)合Python的redis庫，可以快速實(shí)現(xiàn)一個(gè)功能強(qiáng)大、高效的實(shí)時(shí)排行榜系統(tǒng)。

這種排行榜實(shí)現(xiàn)方案非常適合用于在線游戲、社交平臺(tái)等各種應(yīng)用場景。

到此這篇關(guān)于使用Redis實(shí)現(xiàn)實(shí)時(shí)排行榜的示例的文章就介紹到這了,更多相關(guān)Redis 實(shí)時(shí)排行榜內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: