使用Redis实现实时排行榜的示例

更新时间：2025年04月21日 08:33:45 作者：库库林_沙琪马

为了实现一个实时排行榜系统,我们可以使用Redis的有序集合,本文主要介绍了使用Redis实现实时排行榜的示例,具有一定的参考价值,感兴趣的可以了解一下

为了实现一个实时排行榜系统，我们可以使用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 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 案例数据

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)

    # 更新得分（加分操作），如果用户不存在，会将其得分初始化为该值
    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库，可以快速实现一个功能强大、高效的实时排行榜系统。

这种排行榜实现方案非常适合用于在线游戏、社交平台等各种应用场景。

到此这篇关于使用Redis实现实时排行榜的示例的文章就介绍到这了,更多相关Redis 实时排行榜内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Redisson 加锁解锁的实现
本文主要介绍了Redisson 加锁解锁的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2022-08-08
使用redis-cli命令实现redis crud操作过程
线上Redis key需删除但无法直连客户端,可用redis-cli命令操作,步骤包括定位服务、登录、选择数据库及执行DEL命令,适用于无图形工具的环境
2025-08-08
Redis大量数据插入过程
本文介绍Redis处理大数据插入的三种高效方法：redis-cli管道模式、Jedis pipelined()方法及RedisTemplate批量保存,旨在减少通信开销并提升吞吐量
2025-08-08
Redis设置key的过期时间
本文主要介绍了Redis设置key的过期时间，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-01-01
Django使用redis配置缓存的方法
Redis是一个内存数据库由于其性能极高，因此经常作为中间件、缓存使用，缓存某些内容是为了保存昂贵计算的结果，这样就不必在下次执行计算，接下来通过本文给大家分享redis配置缓存的方法，感兴趣的朋友一起看看吧
2021-06-06
Redis 哈希Hash底层数据结构详解
这篇文章主要介绍了Redis 哈希Hash底层数据结构详解的相关资料,需要的朋友可以参考下
2022-08-08
阿里云服务器安装配置redis的方法并且加入到开机启动(推荐)
这篇文章主要介绍了阿里云服务器安装配置redis并且加入到开机启动,需要的朋友可以参考下
2017-12-12
解决Redis的缓存与数据库双写不一致问题
在使用缓存和数据库配合时,常见的CacheAsidePattern模式要求读操作先访问缓存,若缺失再读数据库并更新缓存；写操作则是先写数据库后删除缓存,但这种模式可能导致缓存与数据库间的双写不一致问题
2024-10-10
Redis中ziplist压缩列表的实现
本文主要介绍了Redis中ziplist压缩列表的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-06-06
Redis中5种BitMap应用场景及实现介绍
Redis BitMap是一种高效的位操作数据结构,这种结构在处理海量数据的布尔型状态时尤其高效,下面小编就来和大家简单介绍一下5种它的应用场景及实现方法吧
2025-04-04